kacrab_protocol/
record.rs

1//! Kafka record batch v2 support.
2//!
3//! Record batches are fairly self-contained: [`RecordBatch::encode`] computes
4//! its own CRC32C and applies any compression chosen via the `attributes`
5//! field; [`RecordBatch::decode`] validates the CRC and decompresses before
6//! parsing inner records.
7
8use bytes::{Buf, Bytes, BytesMut};
9
10use crate::primitives::{read_signed_varint, signed_varint_len, write_signed_varint};
11
12pub mod batch;
13pub mod entry;
14pub mod error;
15pub mod header;
16
17pub use self::{
18    batch::{RecordBatch, TimestampType, decode_batches, decode_next_batch},
19    entry::Record,
20    error::{RecordError, RecordErrorKind},
21    header::RecordHeader,
22};
23
24/// Result alias for record-batch operations.
25pub type Result<T> = core::result::Result<T, RecordError>;
26
27/// Maximum number of records permitted in a single batch.
28///
29/// Used to cap `Vec::with_capacity` so a hostile broker cannot trigger a
30/// multi-hundred-GiB allocation (each `Record` is ~100 bytes, so `i32::MAX`
31/// records ≈ 200 GiB) by sending `record_count = i32::MAX`.
32pub const MAX_RECORDS_PER_BATCH: usize = 1_000_000;
33
34fn max_i32_len() -> usize {
35    usize::try_from(i32::MAX).unwrap_or(usize::MAX)
36}
37
38const fn length_overflow(field: &'static str, got: usize, remaining: usize) -> RecordError {
39    RecordError::unknown_offset(RecordErrorKind::LengthOverflow {
40        field,
41        got,
42        remaining,
43    })
44}
45
46fn encode_len(field: &'static str, len: usize) -> Result<i32> {
47    i32::try_from(len).map_err(|_| length_overflow(field, len, max_i32_len()))
48}
49
50pub(super) fn add_encoded_len(field: &'static str, current: usize, addend: usize) -> Result<usize> {
51    current
52        .checked_add(addend)
53        .ok_or_else(|| length_overflow(field, usize::MAX, max_i32_len()))
54}
55
56pub(super) fn bytes_field_len(field: &'static str, bytes: &Bytes) -> Result<usize> {
57    let len = encode_len(field, bytes.len())?;
58    add_encoded_len(field, signed_varint_len(len), bytes.len())
59}
60
61pub(super) fn nullable_bytes_field_len(
62    field: &'static str,
63    bytes: Option<&Bytes>,
64) -> Result<usize> {
65    bytes.map_or_else(
66        || Ok(signed_varint_len(-1)),
67        |bytes| bytes_field_len(field, bytes),
68    )
69}
70
71fn split_exact(buf: &mut Bytes, field: &'static str, len: usize) -> Result<Bytes> {
72    let remaining = buf.remaining();
73    if len > remaining {
74        return Err(length_overflow(field, len, remaining));
75    }
76    Ok(buf.split_to(len))
77}
78
79pub(super) fn read_bytes_field(buf: &mut Bytes, field: &'static str) -> Result<Bytes> {
80    let length = read_signed_varint(buf)?;
81    if length < 0 {
82        return Err(RecordError::unknown_offset(
83            RecordErrorKind::NegativeLength { field, length },
84        ));
85    }
86    let len =
87        usize::try_from(length).map_err(|_| length_overflow(field, usize::MAX, buf.remaining()))?;
88    split_exact(buf, field, len)
89}
90
91pub(super) fn read_nullable_bytes_field(
92    buf: &mut Bytes,
93    field: &'static str,
94) -> Result<Option<Bytes>> {
95    let length = read_signed_varint(buf)?;
96    if length < 0 {
97        return Ok(None);
98    }
99    let len =
100        usize::try_from(length).map_err(|_| length_overflow(field, usize::MAX, buf.remaining()))?;
101    split_exact(buf, field, len).map(Some)
102}
103
104pub(super) fn write_bytes_field(
105    buf: &mut BytesMut,
106    field: &'static str,
107    bytes: &Bytes,
108) -> Result<()> {
109    write_signed_varint(buf, encode_len(field, bytes.len())?);
110    buf.extend_from_slice(bytes);
111    Ok(())
112}
113
114pub(super) fn write_nullable_bytes_field(
115    buf: &mut BytesMut,
116    field: &'static str,
117    bytes: Option<&Bytes>,
118) -> Result<()> {
119    match bytes {
120        None => {
121            write_signed_varint(buf, -1);
122            Ok(())
123        },
124        Some(bytes) => write_bytes_field(buf, field, bytes),
125    }
126}
kacrab_protocol/record.rs

kacrab_protocol/
record.rs
