crates/gunnar-pktline/src/decode.rs
//! Decoding, one-shot over a slice and incrementally over a socket.
use crate::{Error, Packet, Result, MAX_PAYLOAD};
/// The outcome of one [`decode`] attempt.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Decoded {
/// A whole packet, and **how many bytes of the input it used**, always
/// including the 4 header bytes. The caller advances by exactly this much;
/// anything else desyncs the stream.
Packet(Packet, usize),
/// Not enough bytes yet. Read more and call again with a longer buffer.
/// This is not an error: a socket is entitled to split a packet, or even its
/// header, across any number of reads.
Incomplete,
}
/// Decode the packet at the front of `buf`.
///
/// Returns [`Decoded::Incomplete`] when `buf` is a proper prefix of a packet,
/// **for every cut point, including inside the 4-byte header**. Errors are
/// reserved for input that cannot become valid no matter how many more bytes
/// arrive:
///
/// - [`Error::BadLength`]: the header is not 4 hex digits.
/// - [`Error::ReservedLength`]: the header is `0003`.
/// - [`Error::TooLong`]: the header declares more than [`MAX_PAYLOAD`].
///
/// The length check happens **on the header**, before a single payload byte is
/// waited for. A stream that says `ffff` and then stops would otherwise park the
/// connection until a read timeout and surface as a network fault rather than as
/// the malformed frame it is.
pub fn decode(buf: &[u8]) -> Result<Decoded> {
if buf.len() < 4 {
return Ok(Decoded::Incomplete);
}
let head: [u8; 4] = [buf[0], buf[1], buf[2], buf[3]];
let len = parse_len(&head)?;
match len {
0 => return Ok(Decoded::Packet(Packet::Flush, 4)),
1 => return Ok(Decoded::Packet(Packet::Delim, 4)),
2 => return Ok(Decoded::Packet(Packet::ResponseEnd, 4)),
3 => return Err(Error::ReservedLength),
_ => {}
}
// `len` counts its own header, so the payload is `len - 4`. Checking the
// declared size here, not after the bytes arrive, is what keeps a corrupt
// header from being indistinguishable from a slow peer.
let payload_len = len - 4;
if payload_len > MAX_PAYLOAD {
return Err(Error::TooLong(payload_len));
}
if buf.len() < len {
return Ok(Decoded::Incomplete);
}
Ok(Decoded::Packet(Packet::Data(buf[4..len].to_vec()), len))
}
/// 4 ASCII hex digits → the declared total line length.
///
/// Upper and lower case are both accepted, as git's own `packet_length` does.
/// gunnar always *writes* lowercase; being strict on the read side would reject
/// a peer that is within spec, for no gain.
fn parse_len(head: &[u8; 4]) -> Result<usize> {
let mut len = 0usize;
for &b in head {
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,
_ => return Err(Error::BadLength(*head)),
};
len = (len << 4) | nibble as usize;
}
Ok(len)
}
/// An incremental pkt-line reader for a stream that arrives in arbitrary chunks.
///
/// Feed it whatever a read returned with [`push`], take whole packets out with
/// [`next_packet`], and when the framed section is over take the untouched tail
/// with [`into_remainder`].
///
/// ```
/// use gunnar_pktline::{Decoder, Packet};
/// let mut d = Decoder::new();
/// d.push(b"0009hello");
/// d.push(b"0000PACK...");
/// assert_eq!(d.next_packet().unwrap(), Some(Packet::Data(b"hello".to_vec())));
/// assert_eq!(d.next_packet().unwrap(), Some(Packet::Flush));
/// // The flush ended the framed section. Do NOT keep framing past it: what
/// // follows is a raw packfile, and it comes back byte-exact:
/// assert_eq!(d.into_remainder(), b"PACK...".to_vec());
/// ```
///
/// Errors are **sticky by construction**: a failing header is not consumed, so
/// every subsequent call returns the same error rather than resynchronising onto
/// what is, by definition, garbage.
///
/// [`push`]: Decoder::push
/// [`next_packet`]: Decoder::next_packet
/// [`into_remainder`]: Decoder::into_remainder
#[derive(Debug, Default, Clone)]
pub struct Decoder {
buf: Vec<u8>,
/// How much of `buf` has already been handed out as packets. Kept as a
/// cursor instead of draining on every packet: a ref advertisement is tens
/// of thousands of lines, and shifting the tail down after each one is
/// quadratic.
pos: usize,
}
/// Compact once the consumed prefix passes this, and only when it is at least
/// half the buffer. A long stream then cannot grow without bound, while a
/// buffer that is mostly unread data is not memmoved repeatedly for a few bytes.
const COMPACT_THRESHOLD: usize = 64 * 1024;
impl Decoder {
/// A decoder with an empty buffer.
pub fn new() -> Self {
Self::default()
}
/// Append freshly read bytes. Any split point is fine, down to one byte at a
/// time, and a zero-length push is a no-op.
pub fn push(&mut self, bytes: &[u8]) {
self.buf.extend_from_slice(bytes);
}
/// The next whole packet, or `None` if the buffered bytes are a proper
/// prefix of one (or there are none).
///
/// `None` means *not yet*, never *end of stream*. Only a [`Packet::Flush`]
/// carries that meaning, and it is returned as a packet like any other so
/// the caller decides what it ends.
pub fn next_packet(&mut self) -> Result<Option<Packet>> {
match decode(&self.buf[self.pos..])? {
Decoded::Packet(pkt, used) => {
self.pos += used;
self.maybe_compact();
Ok(Some(pkt))
}
Decoded::Incomplete => Ok(None),
}
}
/// Every packet currently complete in the buffer, in order.
///
/// Stops cleanly at the first incomplete packet, leaving its bytes buffered
/// for the next read, so calling this after each socket read yields the
/// same packet sequence as calling [`Decoder::next_packet`] in a loop, no
/// matter where the reads happened to split.
pub fn drain(&mut self) -> Result<Vec<Packet>> {
let mut out = Vec::new();
while let Some(pkt) = self.next_packet()? {
out.push(pkt);
}
Ok(out)
}
/// The buffered bytes that have **not** been handed out as packets.
///
/// This is the raw packfile in `receive-pack`: command list, flush-pkt, then
/// unframed pack bytes, all of which arrive through the same socket and are
/// routinely read into the same buffer. The pack's trailing hash covers every
/// one of these bytes, so they must come back exactly as they arrived: no
/// byte dropped, none duplicated, none reordered.
pub fn remainder(&self) -> &[u8] {
&self.buf[self.pos..]
}
/// [`Decoder::remainder`], taking ownership so the tail can be handed to the
/// pack reader without a copy.
pub fn into_remainder(mut self) -> Vec<u8> {
self.buf.drain(..self.pos);
self.buf
}
/// Number of unconsumed bytes still buffered.
pub fn pending(&self) -> usize {
self.buf.len() - self.pos
}
/// True when nothing is buffered beyond the packets already returned.
pub fn is_empty(&self) -> bool {
self.pending() == 0
}
fn maybe_compact(&mut self) {
if self.pos >= COMPACT_THRESHOLD && self.pos * 2 >= self.buf.len() {
self.buf.drain(..self.pos);
self.pos = 0;
}
}
}