gunnar_sendpack/

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)
}