crates/gunnar-wire/src/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()
);
}
}