crates/gunnar-sendpack/src/driver.rs
//! The client half: pkt-line framing over a blocking stream, and the
//! [`send_pack`] driver that runs one whole push.
//!
//! # The conversation, exactly
//!
//! ```text
//! S→C <old-oid> <refname>\0<capabilities>\n ref advertisement, protocol v0
//! S→C <old-oid> <refname>\n …one line per ref
//! S→C 0000
//! C→S <old> <new> <refname>\0<capabilities>\n the command list
//! C→S <old> <new> <refname>\n …one line per ref being moved
//! C→S 0000
//! C→S <push-option>\n … 0000 only if `push-options` was negotiated
//! C→S PACK… raw, NOT pkt-line framed
//! S→C unpack ok\n report-status
//! S→C ok refs/heads/main\n | ng refs/heads/x <why>\n
//! S→C 0000
//! ```
//!
//! Three details are where implementations go wrong, and each has a test:
//!
//! 1. **The packfile is not framed.** It follows the flush-pkt as raw bytes.
//! Framing it is a corrupt push with a checksum error and no clue pointing
//! at the framing layer.
//! 2. **An all-deletes push sends no pack at all.** Sending an empty pack is
//! also legal, but sending one for a push that introduces no objects is a
//! lie about what the client did.
//! 3. **With `side-band-64k`, the report is muxed inside band 1**, pkt-lines
//! nested inside pkt-lines, and the remote's hook output arrives on band 2.
//! Reading the report without demuxing yields a leading `\x01` on every line
//! and a parse that fails in a way that looks like a protocol error.
use std::io::{Read, Write};
use gix_hash::Kind;
use gix_packetline::blocking_io::encode;
use gix_packetline::decode::{self, Stream};
use gix_packetline::{Channel, PacketLineRef};
use crate::advertisement::{self, Advertisement};
use crate::command::{self, PushCommand};
use crate::error::{Error, Result};
use crate::options::{self, SendPackOptions};
use crate::report::{self, PushReport};
use crate::transport::Transport;
/// Pulls whole pkt-lines out of a byte stream, refilling as needed.
///
/// The one rule that matters: **a short buffer is *incomplete*, never
/// malformed.** A packet split across two reads — including split inside its
/// own four-byte header — is normal on any real socket, and treating it as an
/// error makes a timing-dependent failure that will not reproduce.
/// [`gix_packetline::decode::streaming`] says which of the two it is, and this
/// reader exists to keep asking rather than to guess.
struct PktReader<'a> {
inner: &'a mut dyn Read,
buffer: Vec<u8>,
consumed: usize,
eof: bool,
}
impl<'a> PktReader<'a> {
fn new(inner: &'a mut dyn Read) -> Self {
PktReader {
inner,
buffer: Vec::new(),
consumed: 0,
eof: false,
}
}
/// The next packet as an owned payload description, or `None` at a clean
/// end of stream.
fn next(&mut self) -> Result<Option<Line>> {
loop {
if self.consumed < self.buffer.len() {
match decode::streaming(&self.buffer[self.consumed..]) {
Ok(Stream::Complete {
line,
bytes_consumed,
}) => {
let out = Line::from(line);
self.consumed += bytes_consumed;
// Reclaim once the read cursor has passed most of the
// buffer, rather than on every packet: a report is a
// handful of lines, but the same reader walks a
// sideband-muxed stream that can be megabytes.
if self.consumed > 64 * 1024 {
self.buffer.drain(..self.consumed);
self.consumed = 0;
}
return Ok(Some(out));
}
Ok(Stream::Incomplete { .. }) => {}
Err(err) => return Err(Error::protocol(format!("malformed pkt-line: {err}"))),
}
}
if self.eof {
let pending = self.buffer.len() - self.consumed;
if pending > 0 {
return Err(Error::protocol(format!(
"stream ended mid-packet with {pending} bytes buffered"
)));
}
return Ok(None);
}
let mut chunk = [0u8; 8192];
let n = self.inner.read(&mut chunk)?;
if n == 0 {
self.eof = true;
} else {
self.buffer.extend_from_slice(&chunk[..n]);
}
}
}
/// Every payload up to the next flush-pkt.
///
/// A clean EOF before the flush is accepted and yields what arrived; some
/// servers close instead of flushing after an error, and turning that into
/// a protocol error would replace the remote's diagnosis with our own.
fn payloads_until_flush(&mut self) -> Result<Vec<Vec<u8>>> {
let mut out = Vec::new();
while let Some(line) = self.next()? {
match line {
Line::Flush => return Ok(out),
Line::Data(d) => out.push(d),
// v2-only control packets have no meaning in this v0
// conversation; carrying on past them would desync silently.
other => {
return Err(Error::protocol(format!(
"unexpected control packet in a receive-pack stream: {other:?}"
)))
}
}
}
Ok(out)
}
}
/// One decoded packet, owned so the reader may refill its buffer.
#[derive(Debug)]
enum Line {
Data(Vec<u8>),
Flush,
Delimiter,
ResponseEnd,
}
impl From<PacketLineRef<'_>> for Line {
fn from(line: PacketLineRef<'_>) -> Self {
match line {
PacketLineRef::Data(d) => Line::Data(d.to_vec()),
PacketLineRef::Flush => Line::Flush,
PacketLineRef::Delimiter => Line::Delimiter,
PacketLineRef::ResponseEnd => Line::ResponseEnd,
}
}
}
/// Read `git-receive-pack`'s ref advertisement from `reader`.
pub fn read_advertisement(reader: &mut dyn Read) -> Result<Advertisement> {
let payloads = PktReader::new(reader).payloads_until_flush()?;
advertisement::parse(&payloads)
}
/// Frame `lines` as text pkt-lines and terminate with a flush-pkt.
///
/// *Text* framing is what git uses here: the newline is the framer's, which is
/// why no `lines` function in this crate emits one.
pub fn frame_section(lines: &[Vec<u8>], out: &mut dyn Write) -> Result<()> {
for line in lines {
encode::text_to_write(line, &mut *out)?;
}
encode::flush_to_write(&mut *out)?;
Ok(())
}
/// The command list, framed and flush-terminated.
pub fn encode_command_list(commands: &[PushCommand], caps: &[String]) -> Result<Vec<u8>> {
let mut out = Vec::new();
frame_section(&command::lines(commands, caps)?, &mut out)?;
Ok(out)
}
/// The push-options section, framed and flush-terminated.
pub fn encode_push_options(options: &[String]) -> Result<Vec<u8>> {
let mut out = Vec::new();
frame_section(&command::push_option_lines(options)?, &mut out)?;
Ok(out)
}
/// Read the report from `reader`, demultiplexing sidebands when negotiated.
pub fn read_report(reader: &mut dyn Read, side_band: bool) -> Result<PushReport> {
let mut pkt = PktReader::new(reader);
if !side_band {
let payloads = pkt.payloads_until_flush()?;
return report::parse(&payloads);
}
// With `side-band-64k` the report is pkt-lines *inside* band 1 of
// pkt-lines. Band 1 is accumulated whole and decoded afterwards rather than
// packet-by-packet, because a nested pkt-line may straddle two band-1
// packets and nothing on the wire says so.
let mut band1: Vec<u8> = Vec::new();
let mut progress: Vec<String> = Vec::new();
let mut errors: Vec<String> = Vec::new();
while let Some(line) = pkt.next()? {
match line {
Line::Flush => break,
Line::Data(d) => {
let Some((&band, rest)) = d.split_first() else {
return Err(Error::protocol(
"an empty sideband packet carries no band number",
));
};
match band {
b if b == Channel::Data as u8 => band1.extend_from_slice(rest),
b if b == Channel::Progress as u8 => {
progress.push(String::from_utf8_lossy(rest).into_owned())
}
b if b == Channel::Error as u8 => {
errors.push(String::from_utf8_lossy(rest).into_owned())
}
other => {
return Err(Error::protocol(format!(
"unknown sideband {other} in the push report"
)))
}
}
}
other => {
return Err(Error::protocol(format!(
"unexpected control packet in the push report: {other:?}"
)))
}
}
}
let mut cursor = std::io::Cursor::new(band1);
let mut inner = PktReader::new(&mut cursor);
let lines = inner.payloads_until_flush()?;
let mut out = report::parse(&lines)?;
out.progress = progress;
out.remote_errors = errors;
Ok(out)
}
/// Run one complete send-pack over `transport`, given an advertisement already
/// read from it.
///
/// `write_pack` streams the packfile straight onto the wire; it is `None` for
/// an all-deletes push, which carries no pack. Splitting the advertisement out
/// of this function is what lets the caller decide *what* to push after seeing
/// what the remote has; that decision is the porcelain's, not the wire's.
pub fn send_pack<T, P>(
transport: &mut T,
advertisement: &Advertisement,
commands: &[PushCommand],
local_hash_kind: Kind,
write_pack: Option<P>,
opts: &SendPackOptions,
) -> Result<PushReport>
where
T: Transport,
P: FnOnce(&mut dyn Write) -> Result<()>,
{
let capabilities = options::negotiate(advertisement, commands, local_hash_kind, opts)?;
let side_band = capabilities.iter().any(|c| c == "side-band-64k");
let expects_report = capabilities
.iter()
.any(|c| c == "report-status" || c == "report-status-v2");
let all_deletes = commands.iter().all(PushCommand::is_delete);
if all_deletes && write_pack.is_some() {
return Err(Error::invalid(
"an all-deletes push introduces no objects and must not carry a pack",
));
}
if !all_deletes && write_pack.is_none() {
return Err(Error::invalid(
"a push that creates or updates a ref must carry a pack, even an empty one",
));
}
let commands_bytes = encode_command_list(commands, &capabilities)?;
let options_bytes = if opts.push_options.is_empty() {
Vec::new()
} else {
encode_push_options(&opts.push_options)?
};
{
let (_, writer) = transport.io();
writer.write_all(&commands_bytes)?;
if !options_bytes.is_empty() {
writer.write_all(&options_bytes)?;
}
// The packfile follows the flush-pkt RAW, not pkt-line framed.
if let Some(write_pack) = write_pack {
write_pack(writer)?;
}
writer.flush()?;
}
if !expects_report {
return Ok(PushReport {
unpack: "ok".to_string(),
..PushReport::default()
});
}
let (reader, _) = transport.io();
let mut report = read_report(reader, side_band)?;
// The report is the remote's account of what it did; the command list is
// the only record of what it was asked to do. Reconciling here rather than
// leaving it to each caller is what makes "no line about this ref" a
// failure by construction — a porcelain that forgot to check would report
// a push that moved nothing as a success. See `PushReport::is_ok`.
let commanded: Vec<bstr::BString> = commands.iter().map(|c| c.name.clone()).collect();
report.reconcile(&commanded);
Ok(report)
}