crates/gunnar-http/src/body.rs
//! Turning one HTTP request/response pair into the byte stream the seam wants.
//!
//! # Why this is not a buffer
//!
//! `gunnar_ssh::seam::GitTransport` takes a duplex stream, because that is what
//! an SSH channel is. HTTP has two half-streams instead, and the obvious
//! shortcut — read the request into a `Vec`, serve into another `Vec`, send it
//! — is wrong in both directions and only *looks* fine in a test:
//!
//! * a **clone** response is the packfile. Buffering it means the server holds
//! a whole repository in memory per concurrent clone, and the client sees
//! nothing at all until the last object is written.
//! * a **push** request body is the packfile. Buffering it means the same, for
//! bytes that arrive at whatever speed the client's uplink manages, and
//! `gunnar-wire` already indexes a pack as it streams.
//!
//! So the request body becomes an [`AsyncRead`] and the response body a
//! [`tokio::io::DuplexStream`] that hyper drains as the transport fills it. The
//! duplex is bounded, so a slow client applies backpressure to the pack writer
//! rather than to the heap.
//!
//! # `Content-Encoding: gzip`
//!
//! `git` gzips an rpc request body, but only one it has already buffered — see
//! `http.postBuffer`, whose default is 1 MiB — and never a push, because a
//! packfile is already compressed. That is what makes [`gunzip`] safe as a
//! bounded in-memory step: it is on the small path by construction, and it
//! carries its own ceiling rather than trusting that.
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::Bytes;
use http_body::{Body, Frame};
use hyper::body::Incoming;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
/// How much of a gzipped request body is read before giving up.
///
/// `git`'s own ceiling is `http.postBuffer` (1 MiB by default) and it only
/// gzips what fits in it, so this is that with a wide margin. A body over it is
/// refused rather than inflated: the compressed size says nothing about the
/// inflated one, and "decompress whatever arrives" is a zip bomb with a
/// `Content-Encoding` header on it.
pub const MAX_GZIPPED_REQUEST_BYTES: usize = 16 * 1024 * 1024;
/// The ceiling on the **compressed** bytes of a gzipped request — the one
/// [`MAX_GZIPPED_REQUEST_BYTES`] is not.
///
/// GUN-005. The gzip branch of `POST /<repo>/git-{upload,receive}-pack` chose
/// itself on the `Content-Encoding: gzip` header alone and then ran
/// `incoming.collect()`, which accumulates the WHOLE body with no bound, and
/// only afterwards called [`gunzip`]. Gunzip's ceiling is on the *inflated*
/// size, so the step that actually allocated had none: `Content-Encoding: gzip`
/// plus `Transfer-Encoding: chunked` plus arbitrarily many bytes is remote
/// memory exhaustion at 1:1 bandwidth cost, and the bytes need not even be
/// valid gzip, because the OOM happens before `gunzip` runs. Anonymous, on a
/// public repository, against a process that is meant to run as PID 1 — so the
/// kill takes the whole box — and the admission gate lets 256 transfers in at
/// once.
///
/// Sized as the inflated ceiling plus a 64 KiB allowance rather than equal to
/// it, because deflate's *stored* block is 5 bytes of header per 65 535 bytes
/// of payload plus 18 bytes of gzip header and trailer: incompressible input at
/// the inflated ceiling comes out about 1.3 KiB LARGER than it went in, and a
/// ceiling that refused a body the inflate step would have accepted would be a
/// new bug wearing the fix's clothes.
pub const MAX_GZIPPED_REQUEST_INPUT_BYTES: usize = MAX_GZIPPED_REQUEST_BYTES + 64 * 1024;
/// The response body: whatever the transport writes into the other end.
pub struct Streamed {
reader: Option<tokio::io::DuplexStream>,
/// The largest frame this body will hand hyper, and therefore the amount of
/// pack that may cross the duplex per wakeup.
///
/// **It is the window, not a constant.** Until 2026-08-04 it was a hardcoded
/// 16 KiB while [`HttpConfig::stream_window`](crate::HttpConfig::stream_window)
/// was 256 KiB, so the window bought nothing: the writer could get 256 KiB
/// ahead and the reader still took 16 KiB bites, sixteen wakeups where one
/// would do. That made the *documented* knob — whose own doc says "smaller
/// costs wakeups" — unable to affect wakeups in either direction.
///
/// Every such wakeup is a futex pair: `serve_blocking`'s thread parks in
/// `SyncIoBridge`, a runtime worker wakes it. **Measured on the reference desktop
/// 2026-08-04**, one gunnar per arm, same 8-CPU cgroup budget, the `year`
/// corpus (28 893 objects) over smart HTTP, 8 concurrent clones, four
/// interleaved rounds:
///
/// | | futex syscalls per clone | peak RSS |
/// |---|---|---|
/// | 16 KiB frame | 10 936 · 10 144 · 10 708 · 10 788 | 176 MB |
/// | window-sized frame | 9 032 · 8 268 · 7 307 · 7 882 | 193 MB |
///
/// **−25% syscalls for +10% peak RSS**, and the two sets do not overlap.
/// Both halves are the same trade in opposite directions, which is the point
/// of putting it under one number: an operator for whom the memory ceiling
/// binds can now lower `stream_window` and actually get fewer bytes in
/// flight, which before this they could not.
///
/// It is **not** why gunnar's CPU per request climbs under concurrency.
/// That was localised the same day to where the threads are allowed to run
/// (see `Server::start`'s CPU-budget warning); this is a minority of the
/// same traffic, fixed because it was wrong, not because it was the cause.
///
/// One window-sized read buffer for the body's whole life — its length IS
/// the frame bound — zero-filled once here and overwritten by every read.
///
/// `poll_frame` used to allocate `BytesMut::zeroed(frame)` PER POLL —
/// including a `Pending` poll, which zeroed 256 KiB and threw it away.
/// Profiled on the reference desktop 2026-08-12 (perf on the serving pid, 32 concurrent
/// clones): that zeroing was the process's hottest symbol, 22 % of serve
/// CPU, resolved to `from_elem::<u8>` in `poll_frame` through mimalloc's
/// `mi_page_malloc_zero`. The frame handed to hyper now copies only the
/// bytes a read actually filled, which is bounded by the data, not by the
/// window. (`#![forbid(unsafe_code)]` rules out reading into
/// uninitialised capacity here, so the one-time zero fill stays.)
scratch: Vec<u8>,
}
impl Streamed {
/// A response body and the half the transport writes into.
///
/// `capacity` is the in-flight window. It is the whole of the flow control
/// between the pack writer and the socket, so it wants to be a few network
/// buffers and not a repository.
pub fn pair(capacity: usize) -> (Streamed, tokio::io::DuplexStream) {
// A zero window is a **hang**, not a small buffer: `tokio::io::duplex(0)`
// accepts no byte ever, so the transfer blocks on its first write and
// the client waits for a response that cannot begin. `stream_window` is
// a public field with no floor of its own, so the floor is here, where
// the consequence is. One byte is absurdly slow and it finishes, which
// is the right shape for a misconfiguration: visible, not fatal.
let capacity = capacity.max(1);
let (writer, reader) = tokio::io::duplex(capacity);
(
Streamed {
reader: Some(reader),
scratch: vec![0u8; capacity],
},
writer,
)
}
}
impl Body for Streamed {
type Data = Bytes;
type Error = io::Error;
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Bytes>, io::Error>>> {
let me = self.get_mut();
let Some(reader) = me.reader.as_mut() else {
return Poll::Ready(None);
};
// Into the body's own scratch buffer — see its field doc for the
// profile that moved the per-poll `BytesMut::zeroed` out of here. The
// frame hyper gets copies exactly the filled bytes and no more.
let mut read = ReadBuf::new(&mut me.scratch);
match Pin::new(reader).poll_read(cx, &mut read) {
Poll::Pending => Poll::Pending,
Poll::Ready(Err(err)) => {
me.reader = None;
Poll::Ready(Some(Err(err)))
}
Poll::Ready(Ok(())) => {
let filled = read.filled();
if filled.is_empty() {
// The transport dropped its half: end of body. Not an
// error, and it must not be reported as one, because a
// completed clone ends exactly here.
me.reader = None;
return Poll::Ready(None);
}
Poll::Ready(Some(Ok(Frame::data(Bytes::copy_from_slice(filled)))))
}
}
}
}
/// A request body, read as bytes.
///
/// Either hyper's own streaming [`Incoming`] or a `Vec` that has already been
/// inflated. One type so the caller does not branch on it twice.
pub enum RequestBody {
/// Straight off the socket.
Stream {
/// The remaining frames.
incoming: Incoming,
/// Bytes taken from a frame that did not fit the caller's buffer.
pending: Bytes,
/// Set once the body has ended.
done: bool,
},
/// Already in memory: an inflated `Content-Encoding: gzip` body.
Buffered(io::Cursor<Vec<u8>>),
}
impl RequestBody {
/// Stream the body as it arrives.
pub fn streaming(incoming: Incoming) -> Self {
RequestBody::Stream {
incoming,
pending: Bytes::new(),
done: false,
}
}
/// Serve an already-inflated body.
pub fn buffered(bytes: Vec<u8>) -> Self {
RequestBody::Buffered(io::Cursor::new(bytes))
}
}
impl AsyncRead for RequestBody {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
out: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
match self.get_mut() {
RequestBody::Buffered(cursor) => Pin::new(cursor).poll_read(cx, out),
RequestBody::Stream {
incoming,
pending,
done,
} => loop {
if !pending.is_empty() {
let take = pending.len().min(out.remaining());
out.put_slice(&pending[..take]);
let _ = pending.split_to(take);
return Poll::Ready(Ok(()));
}
if *done {
return Poll::Ready(Ok(()));
}
match Pin::new(&mut *incoming).poll_frame(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(None) => {
*done = true;
return Poll::Ready(Ok(()));
}
Poll::Ready(Some(Err(err))) => {
*done = true;
return Poll::Ready(Err(io::Error::other(err)));
}
Poll::Ready(Some(Ok(frame))) => match frame.into_data() {
// A trailers frame carries no bytes; skip it and read
// on. Returning `Ok(())` here with nothing written
// would look exactly like end-of-body to the caller,
// which would truncate a push at whatever point a
// trailer happened to arrive.
Err(_) => continue,
Ok(data) => *pending = data,
},
}
},
}
}
}
/// A duplex over two independent halves.
///
/// `gunnar_ssh::seam::DuplexStream` is blanket-implemented for anything that is
/// `AsyncRead + AsyncWrite`, and HTTP's two halves are separate objects, so
/// this is what makes an HTTP request satisfy the seam an SSH channel satisfies
/// natively. Nothing here interprets a byte.
pub struct HalfDuplex<R, W> {
read: R,
write: W,
}
impl<R, W> HalfDuplex<R, W> {
/// Join a reader and a writer.
pub fn new(read: R, write: W) -> Self {
HalfDuplex { read, write }
}
}
impl<R: AsyncRead + Unpin, W: Unpin> AsyncRead for HalfDuplex<R, W> {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
out: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().read).poll_read(cx, out)
}
}
impl<R: Unpin, W: AsyncWrite + Unpin> AsyncWrite for HalfDuplex<R, W> {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
data: &[u8],
) -> Poll<io::Result<usize>> {
Pin::new(&mut self.get_mut().write).poll_write(cx, data)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().write).poll_flush(cx)
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().write).poll_shutdown(cx)
}
}
/// A reader that is always at end of file.
///
/// `GET /info/refs` has no request body, and the transport still wants
/// something to read from. Handing it the real (empty) `Incoming` would work
/// too; this says *"there is nothing here"* at the type level, so an
/// advertisement path that started reading would stop rather than wait.
pub struct NoBody;
impl AsyncRead for NoBody {
fn poll_read(
self: Pin<&mut Self>,
_cx: &mut Context<'_>,
_out: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
/// Inflate a `Content-Encoding: gzip` request body.
///
/// Bounded by [`MAX_GZIPPED_REQUEST_BYTES`] on the **inflated** size, which is
/// the number that matters: a few kilobytes of compressed input can name an
/// arbitrarily large one.
pub fn gunzip(compressed: &[u8]) -> io::Result<Vec<u8>> {
use std::io::Read as _;
let mut out = Vec::new();
flate2::read::GzDecoder::new(compressed)
.take(MAX_GZIPPED_REQUEST_BYTES as u64 + 1)
.read_to_end(&mut out)?;
if out.len() > MAX_GZIPPED_REQUEST_BYTES {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"a gzipped request body inflates past the {MAX_GZIPPED_REQUEST_BYTES}-byte ceiling"
),
));
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write as _;
#[test]
fn gunzip_round_trips_and_refuses_a_bomb() {
let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
encoder.write_all(b"0032want abc\n0000").expect("write");
let compressed = encoder.finish().expect("finish");
assert_eq!(gunzip(&compressed).expect("inflate"), b"0032want abc\n0000");
// A body that inflates past the ceiling is refused, and cheaply: this
// input is a few kilobytes.
let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::best());
let chunk = vec![0u8; 1 << 20];
for _ in 0..(MAX_GZIPPED_REQUEST_BYTES / chunk.len() + 2) {
encoder.write_all(&chunk).expect("write");
}
let bomb = encoder.finish().expect("finish");
assert!(
bomb.len() < 128 * 1024,
"the fixture is supposed to be small compressed ({} bytes)",
bomb.len()
);
let err = gunzip(&bomb).expect_err("must refuse");
assert_eq!(err.kind(), io::ErrorKind::InvalidData, "{err}");
}
#[tokio::test]
async fn a_streamed_body_ends_when_the_writer_is_dropped_and_is_not_an_error() {
use http_body_util::BodyExt as _;
let (body, mut writer) = Streamed::pair(4096);
tokio::spawn(async move {
use tokio::io::AsyncWriteExt as _;
writer.write_all(b"0008NAK\n").await.expect("write");
writer.flush().await.expect("flush");
// Dropped here, which is what a finished transfer looks like.
});
let collected = body.collect().await.expect("a clean end, not an error");
assert_eq!(collected.to_bytes().as_ref(), b"0008NAK\n");
}
/// **The window is the frame.** A body given a 256 KiB window must hand a
/// 256 KiB frame out when 256 KiB is waiting, rather than taking bites of
/// its own size and waking the writer for each one.
///
/// Asserted on the FRAME BOUNDARIES and not on the total: every wrong frame
/// size in the world collects to the same bytes, so a `collect()` here would
/// pass on the 16 KiB constant this replaced, on a 1-byte frame, and on any
/// other number a future edit put there.
#[tokio::test]
async fn a_body_hands_out_frames_the_size_of_its_window_not_of_a_constant() {
use http_body_util::BodyExt as _;
const WINDOW: usize = 256 * 1024;
let (mut body, mut writer) = Streamed::pair(WINDOW);
// Fill the window before reading, so what the reader gets back is a
// statement about the reader and not about how fast the writer ran.
tokio::spawn(async move {
use tokio::io::AsyncWriteExt as _;
writer.write_all(&vec![7u8; WINDOW]).await.expect("write");
writer.flush().await.expect("flush");
});
tokio::task::yield_now().await;
let first = body
.frame()
.await
.expect("a body with a full window has a frame")
.expect("and it is not an error");
let n = first.data_ref().expect("a data frame").len();
assert!(
n > 16 * 1024,
"the first frame carried {n} bytes out of a {WINDOW}-byte window. 16 KiB was the \
hardcoded constant this replaced, and at that size a packfile crosses the duplex \
in sixteen times as many wakeups as the window asked for — each one a futex pair."
);
assert!(
n <= WINDOW,
"a frame of {n} bytes is larger than the {WINDOW}-byte window, so the window is no \
longer the bound on what is in flight"
);
}
/// A zero window must not produce a body that reads nothing for ever.
#[tokio::test]
async fn a_zero_window_still_makes_progress_rather_than_hanging() {
use http_body_util::BodyExt as _;
let (body, mut writer) = Streamed::pair(0);
tokio::spawn(async move {
use tokio::io::AsyncWriteExt as _;
let _ = writer.write_all(b"hi").await;
let _ = writer.flush().await;
});
let collected = tokio::time::timeout(std::time::Duration::from_secs(5), body.collect())
.await
.expect("a zero window must not hang the body")
.expect("a clean end");
assert_eq!(collected.to_bytes().as_ref(), b"hi");
}
}