synvael/crates/net/src/codec.rs

// SPDX-License-Identifier: AGPL-3.0-only

//! Length-prefixed `postcard` frame codec.
//!
//! Encodes and decodes one logical protocol message per record on a QUIC stream, using a length prefix so a reader can recover record boundaries from a byte stream.

use crate::error::NetError;

/// The maximum payload length, in bytes, accepted on the control stream (64 KiB), matching the mod-payload cap. Higher-bandwidth tiers such as chunk streaming define their own caps.
pub const MAX_CONTROL_FRAME_LEN: usize = 64 * 1024;

/// The maximum payload length, in bytes, accepted on a chunk stream (1 MiB). A worst-case fully-modified 32³ chunk serializes to roughly 256 KiB as a sparse `ChunkData` (32 768 edits of a varint index plus a `u16` block), so 1 MiB clears the worst case with comfortable margin while still bounding a malicious or corrupt peer's allocation.
pub const MAX_CHUNK_FRAME_LEN: usize = 1024 * 1024;

/// The maximum number of bytes an unsigned LEB128 varint may occupy for a `u64` value (`ceil(64 / 7)`).
const MAX_VARINT_LEN: usize = 10;

/// Appends `value` to `buf` as an unsigned LEB128 varint.
///
/// Each byte carries seven value bits in little-endian group order; the high bit (`0x80`) is a continuation flag set on every byte except the last.
fn write_varint(value: u64, buf: &mut Vec<u8>) {
    let mut remaining = value;
    loop {
        // Extract the low seven bits of the remaining value.
        let mut byte = (remaining & 0x7f) as u8;
        remaining >>= 7;
        if remaining != 0 {
            // Further bytes follow, so mark the continuation bit.
            byte |= 0x80;
        }
        buf.push(byte);
        if remaining == 0 {
            break;
        }
    }
}

/// Reads an unsigned LEB128 varint from the front of `bytes`, returning the decoded value and the number of bytes consumed.
///
/// # Errors
///
/// Returns [`NetError::MalformedVarint`] if the encoding exceeds the ten bytes a `u64` may occupy, or [`NetError::UnexpectedEof`] if the buffer ends while the continuation bit is still set.
fn read_varint(bytes: &[u8]) -> Result<(u64, usize), NetError> {
    let mut value: u64 = 0;
    let mut shift: u32 = 0;
    for (index, &byte) in bytes.iter().enumerate() {
        if index >= MAX_VARINT_LEN {
            return Err(NetError::MalformedVarint);
        }
        // Accumulate the seven payload bits at their little-endian position.
        value |= u64::from(byte & 0x7f) << shift;
        if byte & 0x80 == 0 {
            return Ok((value, index + 1));
        }
        shift += 7;
    }
    // The continuation bit was still set when the buffer ran out.
    Err(NetError::UnexpectedEof)
}

/// Encodes `msg` as a single length-prefixed `postcard` frame into a freshly allocated buffer.
///
/// # Errors
///
/// Returns [`NetError::Postcard`] if `msg` fails to serialize.
fn encode_frame<T: serde::Serialize>(msg: &T) -> Result<Vec<u8>, NetError> {
    let payload = postcard::to_stdvec(msg)?;
    let mut frame = Vec::new();
    write_varint(payload.len() as u64, &mut frame);
    frame.extend_from_slice(&payload);
    Ok(frame)
}

/// Writes one length-prefixed `postcard` frame to a quinn send stream.
///
/// # Errors
///
/// Returns [`NetError::Postcard`] if `msg` fails to serialize, or [`NetError::Write`] if the send stream rejects the bytes.
pub async fn write_frame<T: serde::Serialize>(
    stream: &mut quinn::SendStream,
    msg: &T,
) -> Result<(), NetError> {
    let frame = encode_frame(msg)?;
    stream.write_all(&frame).await?;
    Ok(())
}

/// Reads one length-prefixed `postcard` frame from a quinn recv stream and decodes it.
///
/// # Errors
///
/// Returns [`NetError::MalformedVarint`] if the length prefix is overlong, [`NetError::FrameTooLarge`] if the declared length exceeds `max_len`, [`NetError::Read`] if the stream ends before the frame is complete, or [`NetError::Postcard`] if the payload fails to deserialize.
pub async fn read_frame<T: serde::de::DeserializeOwned>(
    stream: &mut quinn::RecvStream,
    max_len: usize,
) -> Result<T, NetError> {
    // The prefix length is not known in advance, so bytes are pulled one at a time until a byte without the continuation flag is read, then decoded by the shared pure helper.
    let mut prefix = Vec::with_capacity(MAX_VARINT_LEN);
    loop {
        let mut byte = [0u8; 1];
        stream.read_exact(&mut byte).await?;
        prefix.push(byte[0]);
        if byte[0] & 0x80 == 0 {
            break;
        }
        if prefix.len() > MAX_VARINT_LEN {
            return Err(NetError::MalformedVarint);
        }
    }
    let (len, _consumed) = read_varint(&prefix)?;

    // The declared length is validated before allocating the payload buffer.
    let checked_len = check_frame_len(len, max_len)?;
    let mut payload = vec![0u8; checked_len];
    stream.read_exact(&mut payload).await?;
    Ok(postcard::from_bytes(&payload)?)
}

/// Validates a declared frame length against `max_len`.
///
/// # Errors
///
/// Returns [`NetError::FrameTooLarge`] if `len` exceeds `max_len`.
fn check_frame_len(len: u64, max_len: usize) -> Result<usize, NetError> {
    if len > max_len as u64 {
        return Err(NetError::FrameTooLarge { len, max: max_len });
    }
    #[expect(
        clippy::cast_possible_truncation,
        reason = "len <= max_len (usize) checked above"
    )]
    Ok(len as usize)
}

#[cfg(test)]
#[path = "tests/codec.rs"]
mod tests;