use super::{Envelope, EnvelopeError, EnvelopeLimits, Opcode};
use crate::{GroupId, MessageId};
mod batch;
mod encoding;
pub use encoding::Encoding;
mod buffer;
pub use batch::{BatchIter, BatchRecord, PayloadRef, RecordBatch, TinyRecords};
mod indexed;
mod owned;
pub use buffer::RecordBuffer;
pub use indexed::{EncodedParts, IndexedRecords};
pub use owned::{OwnedRecord, OwnedRecordIter, OwnedRecords};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Authority {
pub group_id: GroupId,
pub config_epoch: u64,
pub view: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DataLimits {
pub envelope: EnvelopeLimits,
pub max_records: usize,
pub max_parts: usize,
pub max_record_bytes: usize,
}
impl DataLimits {
#[must_use]
pub fn intersection(self, other: Self) -> Self {
Self {
envelope: EnvelopeLimits {
max_metadata_bytes: self
.envelope
.max_metadata_bytes
.min(other.envelope.max_metadata_bytes),
max_payload_bytes: self
.envelope
.max_payload_bytes
.min(other.envelope.max_payload_bytes),
},
max_records: self.max_records.min(other.max_records),
max_parts: self.max_parts.min(other.max_parts),
max_record_bytes: self.max_record_bytes.min(other.max_record_bytes),
}
}
}
#[cfg(test)]
mod record_limit_tests {
use super::*;
#[test]
fn record_ceiling_is_independent_of_packet_capacity() {
let limits = DataLimits {
envelope: EnvelopeLimits {
max_payload_bytes: 4 * 1024 * 1024,
..EnvelopeLimits::default()
},
..DataLimits::default()
};
let payload = vec![7; 600 * 1024];
let parts = [payload.as_slice()];
let records = [Record {
encoding: crate::data::Encoding::Raw,
message_id: MessageId::new(),
parts: &parts,
}; 2];
assert!(records_size(&records, 0, limits).is_ok());
let large_parts = [payload.as_slice(), payload.as_slice()];
let large = [Record {
encoding: crate::data::Encoding::Raw,
message_id: MessageId::new(),
parts: &large_parts,
}];
assert!(matches!(
records_size(&large, 0, limits),
Err(CodecError::Limit)
));
let mut metadata = Vec::new();
let mut wire_payload = Vec::new();
encode_records(&large, &mut metadata, Some(&mut wire_payload));
assert!(matches!(
decode_records(&metadata, &wire_payload, 0, limits),
Err(CodecError::Limit)
));
}
}
impl Default for DataLimits {
fn default() -> Self {
Self {
envelope: EnvelopeLimits {
max_payload_bytes: 16 * 1024 * 1024,
..EnvelopeLimits::default()
},
max_records: 1000,
max_parts: 4000,
max_record_bytes: 1024 * 1024,
}
}
}
#[derive(Debug, Clone, Copy)]
pub struct Record<'a> {
pub encoding: Encoding,
pub message_id: MessageId,
pub parts: &'a [&'a [u8]],
}
#[derive(Debug, Clone, Copy)]
pub struct Records<'a> {
descriptors: &'a [u8],
payload: &'a [u8],
count: usize,
summary: Option<RecordSummary>,
}
#[derive(Debug, Clone, Copy)]
struct RecordSummary {
parts: usize,
nonzero_message_ids: bool,
}
#[derive(Debug, Clone, Copy)]
pub struct RecordDescriptors<'a> {
framed: &'a [u8],
descriptors: &'a [u8],
count: usize,
payload_bytes: usize,
parts: usize,
nonzero_message_ids: bool,
uniform_part_bytes: Option<u32>,
}
#[derive(Debug, Clone, Copy)]
pub struct RecordDescriptor<'a> {
pub message_id: MessageId,
pub part_lengths: &'a [[u8; 4]],
}
#[derive(Debug, Clone)]
pub struct RecordDescriptorIter<'a> {
descriptors: Cursor<'a>,
remaining: usize,
}
pub(super) fn records_size(
records: &[Record<'_>],
first: u64,
limits: DataLimits,
) -> Result<(usize, usize), CodecError> {
record_sizes(
records
.iter()
.map(|r| (r.message_id, r.encoding, r.parts.iter().copied())),
first,
limits,
)
}
pub(super) fn record_sizes<'a, I, P>(
records: I,
first: u64,
limits: DataLimits,
) -> Result<(usize, usize), CodecError>
where
I: ExactSizeIterator<Item = (MessageId, Encoding, P)>,
P: Iterator<Item = &'a [u8]>,
{
record_count(records.len(), first, limits)?;
let mut metadata = 4_usize;
let mut payload = 0_usize;
let mut parts = 0_usize;
for (_, encoding, record) in records {
metadata = add(metadata, 20 + encoding.metadata_bytes())?;
let before = parts;
let payload_before = payload;
for part in record {
if encoding
.frame_offset(part, limits.max_parts, limits.max_record_bytes)
.is_none()
{
return Err(CodecError::Length);
}
parts = add(parts, 1)?;
count(parts)?;
if parts > limits.max_parts {
return Err(CodecError::Limit);
}
metadata = add(metadata, 4)?;
count(part.len())?;
payload = add(payload, part.len())?;
if payload - payload_before > limits.max_record_bytes {
return Err(CodecError::Limit);
}
}
if !encoding.validate(parts - before, limits.max_parts, limits.max_record_bytes) {
return Err(CodecError::Parts);
}
}
Ok((metadata, payload))
}
pub(super) fn encode_records(
records: &[Record<'_>],
metadata: &mut Vec<u8>,
payload: Option<&mut Vec<u8>>,
) {
write_record_frames(
records
.iter()
.map(|r| (r.message_id, r.encoding, r.parts.iter().copied())),
metadata,
payload,
);
}
pub(super) fn write_records<'a, I, P>(records: I, metadata: &mut Vec<u8>, payload: &mut Vec<u8>)
where
I: ExactSizeIterator<Item = (MessageId, Encoding, P)>,
P: Iterator<Item = &'a [u8]>,
{
write_record_frames(records, metadata, Some(payload));
}
fn write_record_frames<'a, I, P>(records: I, metadata: &mut Vec<u8>, payload: Option<&mut Vec<u8>>)
where
I: ExactSizeIterator<Item = (MessageId, Encoding, P)>,
P: Iterator<Item = &'a [u8]>,
{
metadata.extend_from_slice(&(records.len() as u32).to_be_bytes());
write_record_entries(records, metadata, payload);
}
pub(super) fn write_record_entries<'a, I, P>(
records: I,
metadata: &mut Vec<u8>,
mut payload: Option<&mut Vec<u8>>,
) where
I: Iterator<Item = (MessageId, Encoding, P)>,
P: Iterator<Item = &'a [u8]>,
{
for (id, encoding, parts) in records {
metadata.extend_from_slice(id.as_bytes());
let count_at = metadata.len();
metadata.extend_from_slice(&[0; 4]);
if let Encoding::Lz4 { decoded_bytes } = encoding {
metadata.extend_from_slice(&decoded_bytes.to_be_bytes());
}
let mut count = 0_u32;
for part in parts {
count += 1;
metadata.extend_from_slice(&(part.len() as u32).to_be_bytes());
if let Some(payload) = payload.as_mut() {
payload.extend_from_slice(part);
}
}
metadata[count_at..count_at + 4].copy_from_slice(&(count | encoding.tag()).to_be_bytes());
}
}
pub fn decode_records<'a>(
metadata: &'a [u8],
payload: &'a [u8],
first: u64,
limits: DataLimits,
) -> Result<Records<'a>, CodecError> {
let mut cursor = Cursor(metadata);
let count = cursor.u32()? as usize;
let (parts, nonzero_message_ids) =
validate_record_entries(cursor.0, payload, count, first, limits)?;
Ok(Records {
descriptors: cursor.0,
payload,
count,
summary: Some(RecordSummary {
parts,
nonzero_message_ids,
}),
})
}
pub fn decode_record_descriptors(
metadata: &[u8],
payload_bytes: usize,
first: u64,
limits: DataLimits,
) -> Result<RecordDescriptors<'_>, CodecError> {
let mut cursor = Cursor(metadata);
let count = cursor.u32()? as usize;
let entries = validate_raw_record_entries(cursor.0, count, payload_bytes, first, limits)?;
Ok(RecordDescriptors {
framed: metadata,
descriptors: entries.descriptors,
count,
payload_bytes,
parts: entries.parts,
nonzero_message_ids: entries.nonzero_message_ids,
uniform_part_bytes: entries.uniform_part_bytes,
})
}
pub(super) struct RawRecordEntries<'a> {
pub(super) parts: usize,
pub(super) descriptors: &'a [u8],
pub(super) nonzero_message_ids: bool,
pub(super) uniform_part_bytes: Option<u32>,
}
pub(super) fn validate_raw_record_entries(
descriptors: &[u8],
count: usize,
payload_bytes: usize,
first: u64,
limits: DataLimits,
) -> Result<RawRecordEntries<'_>, CodecError> {
record_count(count, first, limits)?;
if payload_bytes > limits.envelope.max_payload_bytes {
return Err(CodecError::Limit);
}
let mut cursor = Cursor(descriptors);
let mut decoded = 0_usize;
let mut total_parts = 0_usize;
let mut nonzero_message_ids = true;
let mut uniform_part_bytes = None;
for index in 0..count {
let before = decoded;
nonzero_message_ids &= cursor.take(16)? != [0; 16];
let (parts, encoding) = descriptor(&mut cursor)?;
if encoding != Encoding::Raw {
return Err(CodecError::Profile);
}
if parts == 0 {
return Err(CodecError::Parts);
}
total_parts = add(total_parts, parts)?;
if total_parts > limits.max_parts {
return Err(CodecError::Limit);
}
let lengths = cursor.take(parts.checked_mul(4).ok_or(CodecError::Length)?)?;
let single = (parts == 1).then(|| u32::from_be_bytes(lengths.try_into().unwrap()));
if index == 0 {
uniform_part_bytes = single;
} else if uniform_part_bytes != single {
uniform_part_bytes = None;
}
for length in lengths.as_chunks::<4>().0 {
decoded = add(decoded, u32::from_be_bytes(*length) as usize)?;
if decoded > payload_bytes {
return Err(CodecError::Length);
}
if decoded - before > limits.max_record_bytes {
return Err(CodecError::Limit);
}
}
}
if !cursor.0.is_empty() || decoded != payload_bytes {
return Err(CodecError::Length);
}
Ok(RawRecordEntries {
parts: total_parts,
descriptors,
nonzero_message_ids,
uniform_part_bytes,
})
}
impl<'a> RecordDescriptors<'a> {
pub const fn framed(self) -> &'a [u8] {
self.framed
}
pub const fn encoded(self) -> &'a [u8] {
self.descriptors
}
pub const fn len(self) -> usize {
self.count
}
pub const fn is_empty(self) -> bool {
self.count == 0
}
pub const fn payload_bytes(self) -> usize {
self.payload_bytes
}
pub const fn parts(self) -> usize {
self.parts
}
pub const fn nonzero_message_ids(self) -> bool {
self.nonzero_message_ids
}
pub const fn uniform_part_bytes(self) -> Option<u32> {
self.uniform_part_bytes
}
pub fn iter(self) -> RecordDescriptorIter<'a> {
RecordDescriptorIter {
descriptors: Cursor(self.descriptors),
remaining: self.count,
}
}
}
impl<'a> Iterator for RecordDescriptorIter<'a> {
type Item = RecordDescriptor<'a>;
fn next(&mut self) -> Option<Self::Item> {
if self.remaining == 0 {
return None;
}
let message_id = MessageId::from_bytes(
self.descriptors
.array()
.expect("validated record descriptor"),
);
let (parts, encoding) =
descriptor(&mut self.descriptors).expect("validated record descriptor");
debug_assert_eq!(encoding, Encoding::Raw);
let lengths = self
.descriptors
.take(parts * 4)
.expect("validated record lengths")
.as_chunks::<4>()
.0;
self.remaining -= 1;
Some(RecordDescriptor {
message_id,
part_lengths: lengths,
})
}
fn size_hint(&self) -> (usize, Option<usize>) {
(self.remaining, Some(self.remaining))
}
}
impl ExactSizeIterator for RecordDescriptorIter<'_> {}
impl std::iter::FusedIterator for RecordDescriptorIter<'_> {}
pub(super) fn validate_record_entries(
descriptors: &[u8],
payload: &[u8],
count: usize,
first: u64,
limits: DataLimits,
) -> Result<(usize, bool), CodecError> {
record_count(count, first, limits)?;
let mut cursor = Cursor(descriptors);
let mut payload_bytes = 0_usize;
let mut total_parts = 0_usize;
let mut nonzero_message_ids = true;
for _ in 0..count {
let payload_before = payload_bytes;
nonzero_message_ids &= cursor.take(16)? != [0; 16];
let (parts, encoding) = descriptor(&mut cursor)?;
if !encoding.validate(parts, limits.max_parts, limits.max_record_bytes) {
return Err(CodecError::Limit);
}
if parts == 0 {
return Err(CodecError::Parts);
}
total_parts = add(total_parts, parts)?;
if total_parts > limits.max_parts {
return Err(CodecError::Limit);
}
let lengths = cursor.take(parts.checked_mul(4).ok_or(CodecError::Length)?)?;
for length in lengths.as_chunks::<4>().0 {
payload_bytes = add(payload_bytes, u32::from_be_bytes(*length) as usize)?;
if payload_bytes > payload.len() {
return Err(CodecError::Length);
}
if payload_bytes - payload_before > limits.max_record_bytes {
return Err(CodecError::Limit);
}
}
if encoding
.frame_offset(
&payload[payload_before..payload_bytes],
limits.max_parts,
limits.max_record_bytes,
)
.is_none()
{
return Err(CodecError::Length);
}
}
if !cursor.0.is_empty() || payload_bytes != payload.len() {
return Err(CodecError::Length);
}
Ok((total_parts, nonzero_message_ids))
}
impl<'a> Records<'a> {
pub const fn encoded(self) -> (&'a [u8], &'a [u8]) {
(self.descriptors, self.payload)
}
pub fn parts(self) -> usize {
match self.summary {
Some(summary) => summary.parts,
None => self.iter().map(|record| record.parts.len()).sum(),
}
}
pub fn nonzero_message_ids(self) -> bool {
match self.summary {
Some(summary) => summary.nonzero_message_ids,
None => self
.iter()
.all(|record| record.message_id.as_bytes() != &[0; 16]),
}
}
pub const fn descriptor_bytes(self) -> usize {
self.descriptors.len()
}
pub const fn payload_bytes(self) -> usize {
self.payload.len()
}
pub const fn len(self) -> usize {
self.count
}
pub const fn is_empty(self) -> bool {
self.count == 0
}
#[must_use]
pub fn skip(self, count: usize) -> Self {
let mut iter = self.iter();
let skipped = iter.by_ref().take(count).count();
Self {
descriptors: iter.descriptors.0,
payload: iter.payload,
count: self.count - skipped,
summary: None,
}
}
pub fn iter(self) -> RecordIter<'a> {
RecordIter {
descriptors: Cursor(self.descriptors),
payload: self.payload,
remaining: self.count,
}
}
}
#[derive(Debug, Clone)]
pub struct DecodedRecord<'a> {
pub encoding: Encoding,
pub message_id: MessageId,
pub parts: Parts<'a>,
}
#[derive(Debug, Clone)]
pub struct RecordIter<'a> {
descriptors: Cursor<'a>,
payload: &'a [u8],
remaining: usize,
}
impl<'a> Iterator for RecordIter<'a> {
type Item = DecodedRecord<'a>;
fn next(&mut self) -> Option<Self::Item> {
if self.remaining == 0 {
return None;
}
let message_id =
MessageId::from_bytes(self.descriptors.array().expect("validated record ID"));
let (count, encoding) = descriptor(&mut self.descriptors).expect("validated descriptor");
let lengths = self
.descriptors
.take(count * 4)
.expect("validated part lengths");
let bytes = lengths
.as_chunks::<4>()
.0
.iter()
.map(|part| u32::from_be_bytes(*part) as usize)
.sum();
let (payload, rest) = self.payload.split_at(bytes);
self.payload = rest;
self.remaining -= 1;
Some(DecodedRecord {
encoding,
message_id,
parts: Parts {
lengths: lengths.as_chunks::<4>().0.iter(),
payload,
},
})
}
fn size_hint(&self) -> (usize, Option<usize>) {
(self.remaining, Some(self.remaining))
}
}
impl ExactSizeIterator for RecordIter<'_> {}
impl std::iter::FusedIterator for RecordIter<'_> {}
#[derive(Debug, Clone)]
pub struct Parts<'a> {
lengths: std::slice::Iter<'a, [u8; 4]>,
payload: &'a [u8],
}
impl<'a> Iterator for Parts<'a> {
type Item = &'a [u8];
fn next(&mut self) -> Option<Self::Item> {
let length = u32::from_be_bytes(*self.lengths.next()?) as usize;
let (part, rest) = self.payload.split_at(length);
self.payload = rest;
Some(part)
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.lengths.size_hint()
}
}
impl ExactSizeIterator for Parts<'_> {}
impl std::iter::FusedIterator for Parts<'_> {}
impl DoubleEndedIterator for Parts<'_> {
fn next_back(&mut self) -> Option<Self::Item> {
let length = u32::from_be_bytes(*self.lengths.next_back()?) as usize;
let (rest, part) = self.payload.split_at(self.payload.len() - length);
self.payload = rest;
Some(part)
}
}
pub(super) fn command(
envelope: Envelope,
opcode: Opcode,
response: bool,
) -> Result<(), CodecError> {
if envelope.opcode != opcode || envelope.response != response || envelope.request_id.is_none() {
return Err(CodecError::Command);
}
Ok(())
}
pub(super) fn record_count(
records: usize,
first: u64,
limits: DataLimits,
) -> Result<(), CodecError> {
if records == 0 || records > limits.max_records {
return Err(CodecError::Limit);
}
first
.checked_add(u64::from(count(records)?))
.ok_or(CodecError::Length)?;
Ok(())
}
pub(super) fn add(left: usize, right: usize) -> Result<usize, CodecError> {
left.checked_add(right).ok_or(CodecError::Length)
}
pub(super) fn count(value: usize) -> Result<u32, CodecError> {
u32::try_from(value).map_err(|_| CodecError::Length)
}
pub(super) fn capacity(buffer: &Vec<u8>, length: usize) -> Result<(), CodecError> {
if buffer.capacity() < length {
return Err(CodecError::Capacity);
}
Ok(())
}
#[derive(Debug, Clone, Copy)]
pub(super) struct Cursor<'a>(pub(super) &'a [u8]);
impl<'a> Cursor<'a> {
pub(super) fn take(&mut self, count: usize) -> Result<&'a [u8], CodecError> {
let (bytes, rest) = self.0.split_at_checked(count).ok_or(CodecError::Length)?;
self.0 = rest;
Ok(bytes)
}
pub(super) fn array<const N: usize>(&mut self) -> Result<[u8; N], CodecError> {
Ok(self.take(N)?.try_into().expect("checked field"))
}
pub(super) fn byte(&mut self) -> Result<u8, CodecError> {
Ok(self.array::<1>()?[0])
}
pub(super) fn u32(&mut self) -> Result<u32, CodecError> {
Ok(u32::from_be_bytes(self.array()?))
}
pub(super) fn u64(&mut self) -> Result<u64, CodecError> {
Ok(u64::from_be_bytes(self.array()?))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum CodecError {
#[error(transparent)]
Envelope(#[from] EnvelopeError),
#[error("invalid data command envelope")]
Command,
#[error("invalid data identity or epoch")]
Identity,
#[error("unknown completion policy")]
Policy,
#[error("data command exceeds record or part limits")]
Limit,
#[error("record has no payload parts")]
Parts,
#[error("invalid data command length or sequence range")]
Length,
#[error("data output buffer capacity exhausted")]
Capacity,
#[error("invalid data confirmation position")]
Position,
#[error("unsupported reader profile")]
Profile,
}
fn descriptor(cursor: &mut Cursor<'_>) -> Result<(usize, Encoding), CodecError> {
let tagged = cursor.u32()?;
let encoding = match tagged >> 24 {
0 => Encoding::Raw,
1 => Encoding::Lz4 {
decoded_bytes: cursor.u32()?,
},
_ => return Err(CodecError::Profile),
};
Ok(((tagged & 0x00ff_ffff) as usize, encoding))
}