gunnar_wire/

wire_io.rs

//! Reading and writing the framed stream, and the sideband multiplexer.
//!
//! Everything here works on `dyn BufRead` / `dyn Write` and nothing else. That
//! is the transport-agnosticism the crate promises: an SSH channel, a pipe, a
//! `Vec<u8>` and a test fixture are all the same to these functions, so the
//! state machines can be driven from a unit test with no socket in sight.

use std::io::{BufRead, Write};

use gunnar_pktline::{
    decode, Decoded, Packet, BAND_ERROR, BAND_PACK, BAND_PROGRESS, SIDEBAND_CHUNK,
};

use crate::{Error, Result};

/// **How much a transport must buffer under these writers**, and why one
/// sideband packet is the unit that decides it.
///
/// [`SidebandWriter`] hands the transport one packet at a time, and
/// `gunnar_pktline::write_sideband` hands it that packet as **three** calls: a
/// 4-byte length, a 1-byte band, then [`SIDEBAND_CHUNK`] bytes of payload. Put a
/// `BufWriter` under that and its capacity decides how those three become
/// writes on the stream underneath:
///
/// | capacity | writes per 65 520-byte packet | smallest write |
/// |---|---|---|
/// | 8 KiB (`BufWriter::new`) | **2** | **5 bytes** |
/// | this constant | 1 per four packets | 262 080 bytes |
///
/// The 8 KiB case is not a small buffer, it is a **defeated** one: 65 515 bytes
/// exceeds the capacity, so `BufWriter` flushes the 5 bytes it is holding as a
/// write of its own and passes the payload straight through. Every packet
/// therefore costs two writes, one of them five bytes.
///
/// That matters because of what is under the transport. `gunnar-cli`'s
/// `WireTransport` runs this state machine on a `spawn_blocking` thread writing
/// into a `SyncIoBridge` over a `tokio::io::duplex`, so **one write is one
/// hand-off** — a `block_on`, a waker, and, when the reader is parked, a futex
/// pair. `gunnar-http`'s `Streamed` body doc records the other half of the same
/// arithmetic measured 2026-08-04: 7 307–10 936 futex calls per clone of
/// the 28 893-object `year` corpus, and −25% of them from widening the
/// **reader**'s frame alone. The writer was still handing over 64 KiB at a time
/// in two pieces.
///
/// 256 KiB is four whole packets, which is where the ratio stops improving: the
/// duplex window this feeds is 256 KiB (`HttpConfig::stream_window`), so a
/// larger buffer would only split its flush against a window that cannot take
/// it. It is deliberately **not** a copy of that number — a transport buffer and
/// a flow-control window are different things that happen to agree — but it must
/// stay at or above one whole packet or the 5-byte write comes back.
/// `transport_buffer_holds_whole_sideband_packets` asserts that on the applied
/// output: it counts writes, not capacities.
pub const TRANSPORT_BUFFER_BYTES: usize = 256 * 1024;

/// Read exactly one pkt-line, or `None` at a clean end of stream.
///
/// **Exact reads, never buffered read-ahead.** `receive-pack` follows its
/// framed command list with a raw packfile on the very same stream, so a reader
/// that pulled "whatever was available" would swallow the head of the pack and
/// the pack's trailing checksum would then fail over bytes nobody can point at.
/// Reading the 4-byte header and then exactly the declared payload leaves the
/// stream positioned on the first pack byte, which is what
/// [`crate::receive_pack`] relies on.
pub(crate) fn read_packet(input: &mut dyn BufRead) -> Result<Option<Packet>> {
    let mut head = [0u8; 4];
    match read_up_to(input, &mut head)? {
        0 => return Ok(None),
        4 => {}
        _ => return Err(Error::UnexpectedEof("a pkt-line header")),
    }

    // `decode` validates the header (hex digits, the reserved `0003`, the
    // length ceiling) and answers outright for the three control packets,
    // whose whole line is those four bytes. `Incomplete` here therefore means
    // "a valid header for a data packet whose payload has not arrived", which
    // is the only case where the length has to be re-read below.
    match decode(&head)? {
        Decoded::Packet(pkt, _) => return Ok(Some(pkt)),
        Decoded::Incomplete => {}
    }

    let len = line_length(&head);
    let mut line = vec![0u8; len];
    line[..4].copy_from_slice(&head);
    input.read_exact(&mut line[4..]).map_err(|err| {
        if err.kind() == std::io::ErrorKind::UnexpectedEof {
            Error::UnexpectedEof("a pkt-line payload")
        } else {
            Error::Io(err)
        }
    })?;

    match decode(&line)? {
        Decoded::Packet(pkt, used) if used == len => Ok(Some(pkt)),
        _ => Err(Error::protocol(format!(
            "a {len}-byte pkt-line did not decode as one packet"
        ))),
    }
}

/// One pkt-line, refusing a clean end of stream.
pub(crate) fn expect_packet(input: &mut dyn BufRead, what: &'static str) -> Result<Packet> {
    read_packet(input)?.ok_or(Error::UnexpectedEof(what))
}

/// The declared total length of a line whose header `decode` already accepted.
///
/// Four hex digits, which `decode` has necessarily just parsed successfully;
/// it returned `Incomplete` rather than `BadLength`. The parse is repeated here
/// only because `Decoded::Incomplete` carries no length; nothing else in gunnar
/// parses a pkt-line header.
fn line_length(head: &[u8; 4]) -> usize {
    head.iter().fold(0usize, |acc, &b| {
        let nibble = match b {
            b'0'..=b'9' => b - b'0',
            b'a'..=b'f' => b - b'a' + 10,
            b'A'..=b'F' => b - b'A' + 10,
            _ => 0,
        };
        (acc << 4) | nibble as usize
    })
}

fn read_up_to(input: &mut dyn BufRead, buf: &mut [u8]) -> Result<usize> {
    let mut filled = 0;
    while filled < buf.len() {
        match input.read(&mut buf[filled..])? {
            0 => break,
            n => filled += n,
        }
    }
    Ok(filled)
}

/// Read pkt-lines until a flush-pkt, returning the payloads with their trailing
/// newline stripped.
///
/// A clean end of stream before the flush is an error: every section in both
/// protocols is flush-terminated, so an absent flush means a truncated peer.
///
/// `budget` is the caller's ceiling on how many of the peer's bytes this may
/// hold. The loop terminates on a flush-pkt the **peer** chooses to send, so
/// without one the only bound on `lines` is the peer's patience.
pub(crate) fn read_section(
    input: &mut dyn BufRead,
    what: &'static str,
    budget: &mut crate::limits::Budget,
) -> Result<Vec<Vec<u8>>> {
    let mut lines = Vec::new();
    loop {
        match expect_packet(input, what)? {
            Packet::Flush => return Ok(lines),
            Packet::Data(data) => {
                budget.spend(data.len(), what)?;
                let text = data.strip_suffix(b"\n").unwrap_or(&data);
                lines.push(text.to_vec());
            }
            other => {
                return Err(Error::protocol(format!(
                    "unexpected {other:?} while reading {what}"
                )))
            }
        }
    }
}

/// Multiplexes a byte stream onto one sideband, buffering up to a full line.
///
/// The band byte is part of the pkt-line payload, so a full sideband line
/// carries `SIDEBAND_CHUNK` bytes of data and not `MAX_PAYLOAD`; that
/// subtraction is made once, in `gunnar-pktline`, and this type only has to
/// respect the resulting chunk size.
pub(crate) struct SidebandWriter<'a> {
    out: &'a mut dyn Write,
    band: u8,
    buf: Vec<u8>,
}

impl<'a> SidebandWriter<'a> {
    pub(crate) fn new(out: &'a mut dyn Write, band: u8) -> Self {
        SidebandWriter {
            out,
            band,
            buf: Vec::with_capacity(SIDEBAND_CHUNK),
        }
    }

    fn emit(&mut self) -> std::io::Result<()> {
        if self.buf.is_empty() {
            return Ok(());
        }
        gunnar_pktline::write_sideband(&mut self.out, self.band, &self.buf).map_err(as_io)?;
        self.buf.clear();
        Ok(())
    }
}

impl Write for SidebandWriter<'_> {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        if data.is_empty() {
            return Ok(0);
        }
        let room = SIDEBAND_CHUNK - self.buf.len();
        let take = room.min(data.len());
        self.buf.extend_from_slice(&data[..take]);
        if self.buf.len() == SIDEBAND_CHUNK {
            self.emit()?;
        }
        Ok(take)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.emit()?;
        self.out.flush()
    }
}

fn as_io(err: gunnar_pktline::Error) -> std::io::Error {
    match err {
        gunnar_pktline::Error::Io(err) => err,
        other => std::io::Error::other(other.to_string()),
    }
}

/// Where a packfile goes: straight down the wire, or on sideband 1.
///
/// Protocol v2 always sidebands the packfile; protocol v0 only does when the
/// client asked for `side-band-64k`. Both are the same `Write` to the pack
/// writer, so the pack streams either way instead of being buffered whole,
/// which for a clone is the difference between a constant and a repository-sized
/// allocation.
pub(crate) enum PackOutput<'a> {
    Raw(&'a mut dyn Write),
    Sideband(SidebandWriter<'a>),
}

impl<'a> PackOutput<'a> {
    pub(crate) fn new(out: &'a mut dyn Write, sideband: bool) -> Self {
        if sideband {
            PackOutput::Sideband(SidebandWriter::new(out, BAND_PACK))
        } else {
            PackOutput::Raw(out)
        }
    }
}

impl Write for PackOutput<'_> {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        match self {
            PackOutput::Raw(out) => out.write(data),
            PackOutput::Sideband(out) => out.write(data),
        }
    }

    fn flush(&mut self) -> std::io::Result<()> {
        match self {
            PackOutput::Raw(out) => out.flush(),
            PackOutput::Sideband(out) => out.flush(),
        }
    }
}

/// Write one text pkt-line to an unsized writer.
///
/// `gunnar-pktline`'s encoders are generic over `W: Write`, which is right for
/// a leaf crate but means `dyn Write` (the shape every entry point in this
/// crate takes) needs a reborrow at each call. These five wrappers make that
/// reborrow once instead of at ninety call sites.
pub(crate) fn text(out: &mut dyn Write, line: &[u8]) -> Result<()> {
    // The reborrow needs a named binding: `&mut *out` on its own makes
    // inference pick `W = dyn Write`, which is unsized.
    let mut sink: &mut dyn Write = out;
    gunnar_pktline::write_text(&mut sink, line)?;
    Ok(())
}

/// Write one data pkt-line verbatim, adding no newline.
pub(crate) fn data(out: &mut dyn Write, payload: &[u8]) -> Result<()> {
    let mut sink: &mut dyn Write = out;
    gunnar_pktline::write_data(&mut sink, payload)?;
    Ok(())
}

/// Write a flush-pkt.
pub(crate) fn flush(out: &mut dyn Write) -> Result<()> {
    let mut sink: &mut dyn Write = out;
    gunnar_pktline::write_flush(&mut sink)?;
    Ok(())
}

/// Write a delim-pkt.
pub(crate) fn delim(out: &mut dyn Write) -> Result<()> {
    let mut sink: &mut dyn Write = out;
    gunnar_pktline::write_delim(&mut sink)?;
    Ok(())
}

/// Write sideband lines on `band`.
pub(crate) fn sideband(out: &mut dyn Write, band: u8, payload: &[u8]) -> Result<()> {
    let mut sink: &mut dyn Write = out;
    gunnar_pktline::write_sideband(&mut sink, band, payload)?;
    Ok(())
}

/// Send a progress message on band 2, or nothing when progress is suppressed
/// or the client is not reading a sideband.
pub(crate) fn progress(out: &mut dyn Write, enabled: bool, message: &str) -> Result<()> {
    if !enabled || message.is_empty() {
        return Ok(());
    }
    sideband(out, BAND_PROGRESS, message.as_bytes())?;
    Ok(())
}

/// Send a fatal message on band 3. The client aborts on it.
pub(crate) fn sideband_error(out: &mut dyn Write, message: &str) -> Result<()> {
    let mut text = message.to_owned();
    if !text.ends_with('\n') {
        text.push('\n');
    }
    sideband(out, BAND_ERROR, text.as_bytes())?;
    Ok(())
}

/// Send an `ERR <message>` pkt-line, the way a v0 server reports a fatal
/// condition *before* any sideband exists. Public for the transport's
/// first-repository door, which refuses in the client's own protocol after the
/// advertisement and before any sideband.
pub fn err_packet(out: &mut dyn Write, message: &str) -> Result<()> {
    let mut line = String::from("ERR ");
    line.push_str(message);
    text(out, line.as_bytes())?;
    Ok(())
}

#[cfg(test)]
mod transport_buffer_tests {
    use std::io::{BufWriter, Write};

    use super::{PackOutput, TRANSPORT_BUFFER_BYTES};

    /// A sink that remembers how it was called, not what it was given.
    ///
    /// The applied output this file is about is the **sequence of writes**, not
    /// the bytes: the bytes are identical at every buffer size and a test that
    /// compared them would pass at 8 KiB and at 256 KiB alike.
    #[derive(Default)]
    struct Calls {
        sizes: Vec<usize>,
    }

    impl Write for Calls {
        fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
            self.sizes.push(data.len());
            Ok(data.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    /// Push `bytes` of packfile through the real sideband writer over a
    /// `BufWriter` of `capacity`, and report every write the stream underneath
    /// actually saw.
    fn writes_under(capacity: usize, bytes: usize) -> Vec<usize> {
        let payload = vec![0x5Au8; bytes];
        let mut calls = Calls::default();
        {
            let mut buffered = BufWriter::with_capacity(capacity, &mut calls);
            {
                let mut pack = PackOutput::new(&mut buffered, true);
                pack.write_all(&payload)
                    .expect("a counting sink cannot fail");
                pack.flush().expect("a counting sink cannot fail");
            }
            buffered.flush().expect("a counting sink cannot fail");
        }
        calls.sizes
    }

    /// **The write count is the measurement, and 8 KiB is the red arm.**
    ///
    /// One MiB of packfile is sixteen sideband packets. Under the default
    /// `BufWriter::new` capacity each of them costs two writes — a 5-byte one
    /// and a 65 515-byte one — because the payload does not fit and `BufWriter`
    /// dumps its header to get out of the way. Under
    /// [`TRANSPORT_BUFFER_BYTES`] four packets accumulate into one write and no
    /// write is small.
    ///
    /// In `gunnar-cli`'s `WireTransport` each of those writes is a `block_on`
    /// across a `tokio::io::duplex`, so this ratio is a hand-off ratio.
    #[test]
    fn transport_buffer_holds_whole_sideband_packets() {
        const MIB: usize = 1024 * 1024;

        let red = writes_under(8 * 1024, MIB);
        assert!(
            red.iter().any(|n| *n <= 8),
            "the 8 KiB arm is supposed to expose a 5-byte write per sideband packet and did not, \
             so this test is no longer measuring what it says: {red:?}"
        );
        assert!(
            red.len() >= 32,
            "1 MiB is 16 sideband packets and the 8 KiB arm should cost two writes each; it cost \
             {} — the red arm has stopped being red",
            red.len()
        );

        let green = writes_under(TRANSPORT_BUFFER_BYTES, MIB);
        assert!(
            green.iter().all(|n| *n > 8),
            "a buffer of {TRANSPORT_BUFFER_BYTES} bytes still handed the stream a write of {:?} \
             bytes, which is a pkt-line header on its own — the payload no longer fits the \
             buffer",
            green.iter().min()
        );
        assert!(
            green.len() * 4 <= red.len(),
            "{} write(s) at {TRANSPORT_BUFFER_BYTES} bytes against {} at 8 KiB is under the 4x \
             this buffer exists to buy",
            green.len(),
            red.len()
        );
    }
}