You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/17 11:44:29 UTC
(iggy) 02/03: rust side batch rewrite
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch remove_v2_api
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 0ea23986b082b1ec0554b1fa0f0c181b98be7eea
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Mon Aug 17 11:53:25 2026 +0200
rust side batch rewrite
---
core/binary_protocol/src/batch.rs | 629 ++++++++++++
core/binary_protocol/src/error.rs | 19 +
core/binary_protocol/src/lib.rs | 6 +-
core/binary_protocol/src/message_layout.rs | 56 --
core/binary_protocol/src/message_view.rs | 617 ------------
core/binary_protocol/src/requests/messages/mod.rs | 4 +-
.../src/requests/messages/send_messages.rs | 651 ++++---------
.../src/responses/messages/poll_messages.rs | 296 ++++--
core/common/src/traits/binary_impls/messages.rs | 16 +-
core/common/src/types/message/message_view_mut.rs | 117 ---
core/common/src/types/message/mod.rs | 2 -
core/common/src/types/message/polled_messages.rs | 101 +-
core/partitions/src/iggy_partition.rs | 41 +-
core/partitions/src/journal.rs | 12 +-
core/partitions/src/log.rs | 22 +-
core/partitions/src/poll_plan.rs | 4 +-
core/partitions/src/state_transfer.rs | 14 +-
core/server/src/bootstrap.rs | 4 -
core/server/src/http/wire.rs | 46 +-
core/server/src/partition_helpers.rs | 32 +-
core/server/src/responses.rs | 134 +--
core/server/src/segment_recovery.rs | 11 +-
core/server_common/src/in_flight.rs | 100 --
core/server_common/src/indexes_mut.rs | 388 --------
core/server_common/src/lib.rs | 8 -
core/server_common/src/messages_batch_mut.rs | 616 ------------
core/server_common/src/messages_batch_set.rs | 378 -------
.../src/segment_storage/index_reader.rs | 360 +------
.../src/segment_storage/index_writer.rs | 44 -
.../src/segment_storage/messages_reader.rs | 119 +--
.../src/segment_storage/messages_writer.rs | 34 +-
core/server_common/src/segment_storage/mod.rs | 66 +-
core/server_common/src/send_messages.rs | 1029 +++++++-------------
core/simulator/src/client.rs | 26 +-
34 files changed, 1586 insertions(+), 4416 deletions(-)
diff --git a/core/binary_protocol/src/batch.rs b/core/binary_protocol/src/batch.rs
new file mode 100644
index 000000000..57a818957
--- /dev/null
+++ b/core/binary_protocol/src/batch.rs
@@ -0,0 +1,629 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The message batch: the one layout a `SendMessages` body, the replicated
+//! prepare, the persisted segment record, and the poll reply all share.
+//!
+//! ```text
+//! [batch header: 256 bytes][blob: message frames]
+//! frame = [header: 48 bytes][payload][user_headers]
+//! ```
+//!
+//! Producers encode this batch directly on the wire (after the routing
+//! metadata section); the server stamps `partition_id`, `base_offset`, and
+//! `base_timestamp` at ingestion/persistence and replicates the bytes
+//! verbatim; polls serve the stored records back. There is no other message
+//! encoding.
+
+use crate::WireError;
+use std::hash::Hasher;
+use twox_hash::XxHash3_64;
+
+/// Size of the batch header. The remainder past the fields below is reserved
+/// and must be zero.
+pub const BATCH_HEADER_SIZE: usize = 256;
+
+/// Size of a message frame header inside the blob.
+pub const BATCH_MESSAGE_HEADER_SIZE: usize = 48;
+
+/// Byte offset of `batch_checksum` inside the batch header.
+pub const BATCH_CHECKSUM_OFFSET: usize = 40;
+
+/// Byte offset of `message_count` inside the batch header.
+pub const BATCH_MESSAGE_COUNT_OFFSET: usize = 48;
+
+/// Upper bound on a message's `timestamp_delta`: the field is a `u32`
+/// microsecond delta against the batch `origin_timestamp`, so a single batch
+/// spans at most ~71.6 minutes of producer clock.
+pub const MAX_TIMESTAMP_DELTA_MICROS: u64 = u32::MAX as u64;
+
+/// The batch header.
+///
+/// A producer encodes it with `partition_id`, `base_offset`, and
+/// `base_timestamp` zero; the server owns those three fields and stamps them
+/// at ingestion (`partition_id`) and persistence (`base_offset`,
+/// `base_timestamp`, plus the `batch_checksum` recompute).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct BatchHeader {
+ pub partition_id: u64,
+ pub base_offset: u64,
+ pub base_timestamp: u64,
+ pub origin_timestamp: u64,
+ /// Total batch size: `BATCH_HEADER_SIZE` + blob length.
+ pub batch_length: u64,
+ pub batch_checksum: u64,
+ pub message_count: u32,
+}
+
+impl BatchHeader {
+ #[must_use]
+ pub const fn new(
+ partition_id: u64,
+ origin_timestamp: u64,
+ batch_length: u64,
+ message_count: u32,
+ ) -> Self {
+ Self {
+ partition_id,
+ base_offset: 0,
+ base_timestamp: 0,
+ origin_timestamp,
+ batch_length,
+ batch_checksum: 0,
+ message_count,
+ }
+ }
+
+ /// # Errors
+ /// [`WireError::UnexpectedEof`] on a short buffer;
+ /// [`WireError::Validation`] on a `batch_length` smaller than the header.
+ pub fn decode(bytes: &[u8]) -> Result<Self, WireError> {
+ if bytes.len() < BATCH_HEADER_SIZE {
+ return Err(WireError::UnexpectedEof {
+ offset: 0,
+ need: BATCH_HEADER_SIZE,
+ have: bytes.len(),
+ });
+ }
+
+ let batch_length = read_u64(bytes, 32);
+ if batch_length < BATCH_HEADER_SIZE as u64 {
+ return Err(WireError::Validation(std::borrow::Cow::Borrowed(
+ "batch length must cover the batch header",
+ )));
+ }
+
+ Ok(Self {
+ partition_id: read_u64(bytes, 0),
+ base_offset: read_u64(bytes, 8),
+ base_timestamp: read_u64(bytes, 16),
+ origin_timestamp: read_u64(bytes, 24),
+ batch_length,
+ batch_checksum: read_u64(bytes, BATCH_CHECKSUM_OFFSET),
+ message_count: read_u32(bytes, BATCH_MESSAGE_COUNT_OFFSET),
+ })
+ }
+
+ /// # Panics
+ /// Panics if `bytes` is shorter than [`BATCH_HEADER_SIZE`].
+ pub fn encode_into(&self, bytes: &mut [u8]) {
+ assert!(bytes.len() >= BATCH_HEADER_SIZE);
+ bytes[..BATCH_HEADER_SIZE].fill(0);
+ bytes[0..8].copy_from_slice(&self.partition_id.to_le_bytes());
+ bytes[8..16].copy_from_slice(&self.base_offset.to_le_bytes());
+ bytes[16..24].copy_from_slice(&self.base_timestamp.to_le_bytes());
+ bytes[24..32].copy_from_slice(&self.origin_timestamp.to_le_bytes());
+ bytes[32..40].copy_from_slice(&self.batch_length.to_le_bytes());
+ bytes[BATCH_CHECKSUM_OFFSET..BATCH_CHECKSUM_OFFSET + 8]
+ .copy_from_slice(&self.batch_checksum.to_le_bytes());
+ bytes[BATCH_MESSAGE_COUNT_OFFSET..BATCH_MESSAGE_COUNT_OFFSET + 4]
+ .copy_from_slice(&self.message_count.to_le_bytes());
+ }
+
+ /// Total batch size in bytes (header + blob).
+ ///
+ /// # Panics
+ /// Panics if `batch_length` exceeds `usize::MAX`.
+ #[must_use]
+ pub fn total_size(&self) -> usize {
+ usize::try_from(self.batch_length).expect("batch length exceeds usize::MAX")
+ }
+
+ /// # Errors
+ /// [`WireError::Validation`] if `batch_length` does not cover the header
+ /// or exceeds `usize`.
+ pub fn blob_len(&self) -> Result<usize, WireError> {
+ self.batch_length
+ .checked_sub(BATCH_HEADER_SIZE as u64)
+ .and_then(|len| usize::try_from(len).ok())
+ .ok_or(WireError::Validation(std::borrow::Cow::Borrowed(
+ "batch length must cover the batch header",
+ )))
+ }
+
+ #[must_use]
+ pub fn checksum_for_blob(&self, blob: &[u8]) -> u64 {
+ calculate_batch_checksum(self, blob)
+ }
+}
+
+/// A decoded batch borrowing its blob.
+#[derive(Debug, Clone, Copy)]
+pub struct BatchRef<'a> {
+ pub header: BatchHeader,
+ blob: &'a [u8],
+}
+
+impl<'a> BatchRef<'a> {
+ #[must_use]
+ pub const fn new(header: BatchHeader, blob: &'a [u8]) -> Self {
+ Self { header, blob }
+ }
+
+ #[must_use]
+ pub const fn iter(&self) -> BatchIterator<'a> {
+ BatchIterator {
+ blob: self.blob,
+ position: 0,
+ }
+ }
+
+ #[must_use]
+ pub const fn iter_with_offsets(&self) -> BatchIteratorWithOffsets<'a> {
+ BatchIteratorWithOffsets {
+ blob: self.blob,
+ position: 0,
+ }
+ }
+
+ #[must_use]
+ pub const fn blob(&self) -> &'a [u8] {
+ self.blob
+ }
+
+ #[must_use]
+ pub const fn message_count(&self) -> u32 {
+ self.header.message_count
+ }
+}
+
+impl<'a> IntoIterator for &BatchRef<'a> {
+ type Item = BatchMessageView<'a>;
+ type IntoIter = BatchIterator<'a>;
+
+ fn into_iter(self) -> Self::IntoIter {
+ self.iter()
+ }
+}
+
+/// A message frame header inside the blob.
+#[derive(Debug, Clone, Copy)]
+pub struct BatchMessageHeader {
+ /// `XxHash3_64` over `frame[8..48] || payload || user_headers`.
+ pub checksum: u64,
+ pub id: u128,
+ pub offset_delta: u32,
+ /// Microsecond delta from [`BatchHeader::origin_timestamp`].
+ pub timestamp_delta: u32,
+ pub user_headers_length: u32,
+ pub payload_length: u32,
+}
+
+impl BatchMessageHeader {
+ /// # Errors
+ /// [`WireError::UnexpectedEof`] on a short buffer; [`WireError::Validation`]
+ /// on nonzero reserved bytes.
+ pub fn decode(bytes: &[u8]) -> Result<Self, WireError> {
+ if bytes.len() < BATCH_MESSAGE_HEADER_SIZE {
+ return Err(WireError::UnexpectedEof {
+ offset: 0,
+ need: BATCH_MESSAGE_HEADER_SIZE,
+ have: bytes.len(),
+ });
+ }
+
+ if read_u64(bytes, 40) != 0 {
+ return Err(WireError::Validation(std::borrow::Cow::Borrowed(
+ "message frame reserved bytes must be zero",
+ )));
+ }
+
+ Ok(Self {
+ checksum: read_u64(bytes, 0),
+ id: read_u128(bytes, 8),
+ offset_delta: read_u32(bytes, 24),
+ timestamp_delta: read_u32(bytes, 28),
+ user_headers_length: read_u32(bytes, 32),
+ payload_length: read_u32(bytes, 36),
+ })
+ }
+
+ /// Frame size: header + payload + user headers.
+ #[must_use]
+ pub const fn total_size(&self) -> usize {
+ BATCH_MESSAGE_HEADER_SIZE + self.user_headers_length as usize + self.payload_length as usize
+ }
+}
+
+/// A message frame view borrowing payload and user headers from the blob.
+#[derive(Debug, Clone, Copy)]
+pub struct BatchMessageView<'a> {
+ pub header: BatchMessageHeader,
+ pub user_headers: &'a [u8],
+ pub payload: &'a [u8],
+}
+
+pub struct BatchIterator<'a> {
+ blob: &'a [u8],
+ position: usize,
+}
+
+impl<'a> Iterator for BatchIterator<'a> {
+ type Item = BatchMessageView<'a>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ if self.position >= self.blob.len() {
+ return None;
+ }
+
+ let header = BatchMessageHeader::decode(&self.blob[self.position..]).ok()?;
+ let start = self.position + BATCH_MESSAGE_HEADER_SIZE;
+ let payload_end = start + header.payload_length as usize;
+ let headers_end = payload_end + header.user_headers_length as usize;
+ let payload = self.blob.get(start..payload_end)?;
+ let user_headers = self.blob.get(payload_end..headers_end)?;
+ self.position += header.total_size();
+ Some(BatchMessageView {
+ header,
+ user_headers,
+ payload,
+ })
+ }
+}
+
+/// A frame view plus its byte range inside the blob.
+#[derive(Debug, Clone, Copy)]
+pub struct BatchMessageViewWithOffsets<'a> {
+ pub message: BatchMessageView<'a>,
+ pub start: usize,
+ pub end: usize,
+}
+
+pub struct BatchIteratorWithOffsets<'a> {
+ blob: &'a [u8],
+ position: usize,
+}
+
+impl<'a> Iterator for BatchIteratorWithOffsets<'a> {
+ type Item = BatchMessageViewWithOffsets<'a>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ if self.position >= self.blob.len() {
+ return None;
+ }
+
+ let start = self.position;
+ let header = BatchMessageHeader::decode(&self.blob[self.position..]).ok()?;
+ let message_start = self.position + BATCH_MESSAGE_HEADER_SIZE;
+ let payload_end = message_start + header.payload_length as usize;
+ let headers_end = payload_end + header.user_headers_length as usize;
+ let payload = self.blob.get(message_start..payload_end)?;
+ let user_headers = self.blob.get(payload_end..headers_end)?;
+ self.position += header.total_size();
+ Some(BatchMessageViewWithOffsets {
+ message: BatchMessageView {
+ header,
+ user_headers,
+ payload,
+ },
+ start,
+ end: self.position,
+ })
+ }
+}
+
+/// How much of a batch record [`decode_batch_slice_with`] proves before returning it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BatchIntegrity {
+ /// Re-hash the batch and reject it unless it matches its own `batch_checksum`.
+ Verify,
+ /// Check the framing only, and hand back whatever it describes. The caller is
+ /// accepting bytes that may not be the ones written.
+ LayoutOnly,
+}
+
+/// Decode one batch record (`[256B header][blob]`), verifying the batch
+/// checksum and every per-message checksum.
+///
+/// `body` may extend past the batch: readers walking a stream of records hand
+/// in the rest of the buffer and step by `batch_length`. Callers whose buffer
+/// is meant to BE the batch must reject the surplus themselves.
+///
+/// # Errors
+/// [`WireError`] on a short or self-inconsistent record, and
+/// [`WireError::InvalidBatchChecksum`] / [`WireError::InvalidMessageChecksum`]
+/// on an integrity mismatch.
+pub fn decode_batch_slice(body: &[u8]) -> Result<BatchRef<'_>, WireError> {
+ decode_batch_slice_with(body, BatchIntegrity::Verify)
+}
+
+/// [`decode_batch_slice`] with the integrity level chosen by the caller.
+///
+/// Layout checks are not optional either way: a short or self-inconsistent
+/// record is rejected regardless, because the caller would otherwise index
+/// past it.
+///
+/// # Errors
+/// See [`decode_batch_slice`]; [`BatchIntegrity::LayoutOnly`] skips only the
+/// checksum comparison.
+pub fn decode_batch_slice_with(
+ body: &[u8],
+ integrity: BatchIntegrity,
+) -> Result<BatchRef<'_>, WireError> {
+ let header = BatchHeader::decode(body)?;
+ let blob_len = header.blob_len()?;
+ if body.len() < header.total_size() {
+ return Err(WireError::UnexpectedEof {
+ offset: 0,
+ need: header.total_size(),
+ have: body.len(),
+ });
+ }
+
+ let blob = &body[BATCH_HEADER_SIZE..BATCH_HEADER_SIZE + blob_len];
+ let batch = BatchRef { header, blob };
+ match integrity {
+ BatchIntegrity::Verify => {
+ let expected_checksum = verify_and_recompute_batch_checksum(&batch)?;
+ if header.batch_checksum != expected_checksum {
+ return Err(WireError::InvalidBatchChecksum {
+ stored: header.batch_checksum,
+ computed: expected_checksum,
+ base_offset: header.base_offset,
+ });
+ }
+ }
+ BatchIntegrity::LayoutOnly => validate_batch_layout(&batch)?,
+ }
+
+ Ok(batch)
+}
+
+/// Batch checksum: streaming `XxHash3_64` over the six batch header meta
+/// fields followed by each message's stored 8-byte checksum field in message
+/// order - NOT the message bodies.
+///
+/// Bodies are bound only transitively: each per-message checksum already covers
+/// `frame[8..48] || payload || user_headers`, so hashing the checksum fields
+/// binds every body byte IFF a reader also re-verifies the per-message
+/// checksums. Stamping hashes `N * 8` bytes instead of the whole blob;
+/// validating decoders pay the one body pass as the per-message verify in
+/// [`verify_and_recompute_batch_checksum`], which hashes the checksum-field
+/// bytes in the same order so its recompute matches a compute here.
+///
+/// Assumes a well-formed blob whose frames tile exactly; every compute site
+/// builds the blob and satisfies this.
+#[must_use]
+pub fn calculate_batch_checksum(header: &BatchHeader, blob: &[u8]) -> u64 {
+ let mut hasher = XxHash3_64::new();
+ write_batch_header_fields(&mut hasher, header);
+ let batch = BatchRef {
+ header: *header,
+ blob,
+ };
+ for framed in batch.iter_with_offsets() {
+ hasher.write(&blob[framed.start..framed.start + 8]);
+ }
+ hasher.finish()
+}
+
+fn write_batch_header_fields(hasher: &mut XxHash3_64, header: &BatchHeader) {
+ hasher.write(&header.partition_id.to_le_bytes());
+ hasher.write(&header.base_offset.to_le_bytes());
+ hasher.write(&header.base_timestamp.to_le_bytes());
+ hasher.write(&header.origin_timestamp.to_le_bytes());
+ hasher.write(&header.batch_length.to_le_bytes());
+ hasher.write(&header.message_count.to_le_bytes());
+}
+
+/// Verify every per-message checksum in `batch` and return the recomputed
+/// batch checksum (see [`calculate_batch_checksum`]) from a single frame walk.
+///
+/// The per-message pass is the equal-integrity half of the scheme: the batch
+/// value binds bodies only through the checksum fields, so a validating decode
+/// must re-verify each message here or body corruption that leaves the
+/// checksum field intact would pass. This is the one full-body pass a
+/// validating decode pays; the caller then compares the returned value against
+/// the stored `batch_checksum`.
+///
+/// # Errors
+/// [`WireError::InvalidMessageChecksum`] on the first per-message mismatch;
+/// [`WireError::Validation`] if the frames do not tile `message_count` exactly.
+pub fn verify_and_recompute_batch_checksum(batch: &BatchRef<'_>) -> Result<u64, WireError> {
+ let blob = batch.blob();
+ let mut hasher = XxHash3_64::new();
+ write_batch_header_fields(&mut hasher, &batch.header);
+ let mut verified = 0u32;
+ let mut covered = 0usize;
+ for framed in batch.iter_with_offsets() {
+ // Cover (`frame[8..48] || payload || user_headers`) hashed raw from the
+ // blob, byte-exact with the encoder's, so a flipped body byte fails even
+ // when the stored checksum field is left intact.
+ let stored = framed.message.header.checksum;
+ let expected = XxHash3_64::oneshot(&blob[framed.start + 8..framed.end]);
+ if expected != stored {
+ return Err(WireError::InvalidMessageChecksum {
+ stored,
+ computed: expected,
+ offset: batch.header.base_offset + u64::from(framed.message.header.offset_delta),
+ });
+ }
+ hasher.write(&blob[framed.start..framed.start + 8]);
+ verified += 1;
+ covered = framed.end;
+ }
+ if verified != batch.message_count() || covered != blob.len() {
+ return Err(WireError::Validation(std::borrow::Cow::Borrowed(
+ "batch frames do not tile message_count exactly",
+ )));
+ }
+ Ok(hasher.finish())
+}
+
+/// Layout-only twin of [`verify_and_recompute_batch_checksum`]: prove the
+/// frames tile `message_count` exactly without touching any checksum.
+///
+/// # Errors
+/// [`WireError::Validation`] if the frames do not tile `message_count` exactly.
+fn validate_batch_layout(batch: &BatchRef<'_>) -> Result<(), WireError> {
+ let blob = batch.blob();
+ let mut counted = 0u32;
+ let mut covered = 0usize;
+ for framed in batch.iter_with_offsets() {
+ counted += 1;
+ covered = framed.end;
+ }
+ if counted != batch.message_count() || covered != blob.len() {
+ return Err(WireError::Validation(std::borrow::Cow::Borrowed(
+ "batch frames do not tile message_count exactly",
+ )));
+ }
+ Ok(())
+}
+
+fn read_u32(bytes: &[u8], offset: usize) -> u32 {
+ u32::from_le_bytes(bytes[offset..offset + 4].try_into().expect("4-byte slice"))
+}
+
+fn read_u64(bytes: &[u8], offset: usize) -> u64 {
+ u64::from_le_bytes(bytes[offset..offset + 8].try_into().expect("8-byte slice"))
+}
+
+fn read_u128(bytes: &[u8], offset: usize) -> u128 {
+ u128::from_le_bytes(
+ bytes[offset..offset + 16]
+ .try_into()
+ .expect("16-byte slice"),
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn frame(id: u128, offset_delta: u32, timestamp_delta: u32, payload: &[u8]) -> Vec<u8> {
+ let mut bytes = vec![0u8; BATCH_MESSAGE_HEADER_SIZE];
+ bytes[8..24].copy_from_slice(&id.to_le_bytes());
+ bytes[24..28].copy_from_slice(&offset_delta.to_le_bytes());
+ bytes[28..32].copy_from_slice(×tamp_delta.to_le_bytes());
+ bytes[36..40].copy_from_slice(&(payload.len() as u32).to_le_bytes());
+ bytes.extend_from_slice(payload);
+ let checksum = XxHash3_64::oneshot(&bytes[8..]);
+ bytes[0..8].copy_from_slice(&checksum.to_le_bytes());
+ bytes
+ }
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn batch_bytes(frames: &[Vec<u8>]) -> Vec<u8> {
+ let blob: Vec<u8> = frames.concat();
+ let mut header = BatchHeader::new(
+ 7,
+ 1_000,
+ (BATCH_HEADER_SIZE + blob.len()) as u64,
+ frames.len() as u32,
+ );
+ header.batch_checksum = calculate_batch_checksum(&header, &blob);
+ let mut bytes = vec![0u8; BATCH_HEADER_SIZE];
+ header.encode_into(&mut bytes);
+ bytes.extend_from_slice(&blob);
+ bytes
+ }
+
+ #[test]
+ fn header_roundtrip() {
+ let mut header = BatchHeader::new(1, 2, 300, 4);
+ header.base_offset = 10;
+ header.base_timestamp = 20;
+ header.batch_checksum = 30;
+ let mut bytes = vec![0u8; BATCH_HEADER_SIZE];
+ header.encode_into(&mut bytes);
+ let decoded = BatchHeader::decode(&bytes).unwrap();
+ assert_eq!(decoded, header);
+ }
+
+ #[test]
+ fn decode_verifies_checksums() {
+ let bytes = batch_bytes(&[frame(1, 0, 0, b"first"), frame(2, 1, 5, b"second")]);
+ let batch = decode_batch_slice(&bytes).unwrap();
+ assert_eq!(batch.message_count(), 2);
+ let views: Vec<_> = batch.iter().collect();
+ assert_eq!(views[0].payload, b"first");
+ assert_eq!(views[1].payload, b"second");
+ assert_eq!(views[1].header.offset_delta, 1);
+ }
+
+ #[test]
+ fn decode_rejects_flipped_body_byte() {
+ let mut bytes = batch_bytes(&[frame(1, 0, 0, b"payload")]);
+ let last = bytes.len() - 1;
+ bytes[last] ^= 0xFF;
+ assert!(matches!(
+ decode_batch_slice(&bytes),
+ Err(WireError::InvalidMessageChecksum { .. })
+ ));
+ }
+
+ #[test]
+ fn decode_rejects_flipped_batch_checksum() {
+ let mut bytes = batch_bytes(&[frame(1, 0, 0, b"payload")]);
+ bytes[BATCH_CHECKSUM_OFFSET] ^= 0xFF;
+ assert!(matches!(
+ decode_batch_slice(&bytes),
+ Err(WireError::InvalidBatchChecksum { .. })
+ ));
+ }
+
+ #[test]
+ fn layout_only_accepts_zero_checksums() {
+ let mut bytes = batch_bytes(&[frame(1, 0, 0, b"payload")]);
+ bytes[BATCH_CHECKSUM_OFFSET..BATCH_CHECKSUM_OFFSET + 8].fill(0);
+ let batch = decode_batch_slice_with(&bytes, BatchIntegrity::LayoutOnly).unwrap();
+ assert_eq!(batch.message_count(), 1);
+ }
+
+ #[test]
+ fn decode_rejects_miscounted_batch() {
+ let mut bytes = batch_bytes(&[frame(1, 0, 0, b"payload")]);
+ bytes[BATCH_MESSAGE_COUNT_OFFSET..BATCH_MESSAGE_COUNT_OFFSET + 4]
+ .copy_from_slice(&2u32.to_le_bytes());
+ assert!(decode_batch_slice_with(&bytes, BatchIntegrity::LayoutOnly).is_err());
+ }
+
+ #[test]
+ fn decode_rejects_truncated_batch() {
+ let bytes = batch_bytes(&[frame(1, 0, 0, b"payload")]);
+ assert!(decode_batch_slice(&bytes[..bytes.len() - 1]).is_err());
+ }
+
+ #[test]
+ fn decode_rejects_nonzero_reserved() {
+ let mut bytes = batch_bytes(&[frame(1, 0, 0, b"payload")]);
+ bytes[BATCH_HEADER_SIZE + 40] = 1;
+ assert!(decode_batch_slice_with(&bytes, BatchIntegrity::LayoutOnly).is_err());
+ }
+}
diff --git a/core/binary_protocol/src/error.rs b/core/binary_protocol/src/error.rs
index 82d5cb2b5..208b15c3f 100644
--- a/core/binary_protocol/src/error.rs
+++ b/core/binary_protocol/src/error.rs
@@ -49,4 +49,23 @@ pub enum WireError {
#[error("validation failed: {0}")]
Validation(Cow<'static, str>),
+
+ #[error(
+ "invalid batch checksum: stored {stored}, computed {computed}, base offset {base_offset}"
+ )]
+ InvalidBatchChecksum {
+ stored: u64,
+ computed: u64,
+ base_offset: u64,
+ },
+
+ #[error("invalid message checksum: stored {stored}, computed {computed}, offset {offset}")]
+ InvalidMessageChecksum {
+ stored: u64,
+ computed: u64,
+ offset: u64,
+ },
+
+ #[error("message timestamp delta {0} exceeds the batch maximum")]
+ InvalidMessageTimestampDelta(u64),
}
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index fa376467e..7ec2ddd93 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -55,14 +55,13 @@
//! [`StartViewHeader`]
//! - Dispatch: [`GenericHeader`] for type-erased initial parsing
+pub mod batch;
pub mod codec;
pub mod codes;
pub mod consensus;
pub mod dispatch;
pub mod error;
pub mod framing;
-pub mod message_layout;
-pub mod message_view;
pub mod namespace;
pub mod primitives;
pub mod requests;
@@ -84,9 +83,6 @@ pub use consensus::{
pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
pub use error::WireError;
pub use framing::{RequestFrame, ResponseFrame, STATUS_OK};
-pub use message_view::{
- WireMessageIterator, WireMessageIteratorMut, WireMessageView, WireMessageViewMut,
-};
pub use primitives::ack_level::AckLevel;
pub use primitives::consumer::{KIND_CONSUMER_GROUP, WireConsumer};
pub use primitives::identifier::{MAX_WIRE_NAME_LENGTH, WireIdentifier, WireName};
diff --git a/core/binary_protocol/src/message_layout.rs b/core/binary_protocol/src/message_layout.rs
deleted file mode 100644
index 2858ec5e2..000000000
--- a/core/binary_protocol/src/message_layout.rs
+++ /dev/null
@@ -1,56 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-//! 64-byte message frame layout constants.
-//!
-//! This module is the single source of truth for the on-wire message
-//! header layout shared between encoders, decoders, and zero-copy views.
-//!
-//! ```text
-//! [checksum:8][id:16][offset:8][timestamp:8][origin_timestamp:8]
-//! [user_headers_length:4][payload_length:4][reserved:8]
-//! ```
-
-/// Fixed-size message header on the wire (64 bytes).
-pub const WIRE_MESSAGE_HEADER_SIZE: usize = 64;
-
-/// Fixed-size index entry per message (16 bytes).
-///
-/// Layout: `[0x00000000:4][cumulative_size:u32_le:4][0x0000000000000000:8]`
-pub const WIRE_MESSAGE_INDEX_SIZE: usize = 16;
-
-// Field offsets within the 64-byte message header.
-pub const MSG_CHECKSUM_OFFSET: usize = 0;
-pub const MSG_ID_OFFSET: usize = 8;
-pub const MSG_OFFSET_OFFSET: usize = 24;
-pub const MSG_TIMESTAMP_OFFSET: usize = 32;
-pub const MSG_ORIGIN_TIMESTAMP_OFFSET: usize = 40;
-pub const MSG_USER_HEADERS_LEN_OFFSET: usize = 48;
-pub const MSG_PAYLOAD_LEN_OFFSET: usize = 52;
-pub const MSG_RESERVED_OFFSET: usize = 56;
-
-// Compile-time verification that field offsets are contiguous and sum to the header size.
-const _: () = {
- assert!(MSG_ID_OFFSET == MSG_CHECKSUM_OFFSET + 8);
- assert!(MSG_OFFSET_OFFSET == MSG_ID_OFFSET + 16);
- assert!(MSG_TIMESTAMP_OFFSET == MSG_OFFSET_OFFSET + 8);
- assert!(MSG_ORIGIN_TIMESTAMP_OFFSET == MSG_TIMESTAMP_OFFSET + 8);
- assert!(MSG_USER_HEADERS_LEN_OFFSET == MSG_ORIGIN_TIMESTAMP_OFFSET + 8);
- assert!(MSG_PAYLOAD_LEN_OFFSET == MSG_USER_HEADERS_LEN_OFFSET + 4);
- assert!(MSG_RESERVED_OFFSET == MSG_PAYLOAD_LEN_OFFSET + 4);
- assert!(MSG_RESERVED_OFFSET + 8 == WIRE_MESSAGE_HEADER_SIZE);
-};
diff --git a/core/binary_protocol/src/message_view.rs b/core/binary_protocol/src/message_view.rs
deleted file mode 100644
index 3159dac5b..000000000
--- a/core/binary_protocol/src/message_view.rs
+++ /dev/null
@@ -1,617 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-//! Zero-copy view types for message frames on the wire.
-//!
-//! These types borrow an existing buffer and provide typed access to
-//! message header fields and payload data without copying.
-
-use crate::error::WireError;
-use crate::message_layout::{
- MSG_CHECKSUM_OFFSET, MSG_ID_OFFSET, MSG_OFFSET_OFFSET, MSG_ORIGIN_TIMESTAMP_OFFSET,
- MSG_PAYLOAD_LEN_OFFSET, MSG_TIMESTAMP_OFFSET, MSG_USER_HEADERS_LEN_OFFSET,
- WIRE_MESSAGE_HEADER_SIZE,
-};
-use std::borrow::Cow;
-
-// Private helpers for infallible reads on validated buffers
-
-#[inline]
-fn u32_at(buf: &[u8], offset: usize) -> u32 {
- u32::from_le_bytes(
- buf[offset..offset + 4]
- .try_into()
- .expect("slice is exactly 4 bytes"),
- )
-}
-
-#[inline]
-fn u64_at(buf: &[u8], offset: usize) -> u64 {
- u64::from_le_bytes(
- buf[offset..offset + 8]
- .try_into()
- .expect("slice is exactly 8 bytes"),
- )
-}
-
-#[inline]
-fn u128_at(buf: &[u8], offset: usize) -> u128 {
- u128::from_le_bytes(
- buf[offset..offset + 16]
- .try_into()
- .expect("slice is exactly 16 bytes"),
- )
-}
-
-/// Validate a message frame buffer and return `(total_size, payload_len, user_headers_len)`.
-fn validate_frame(buf: &[u8]) -> Result<(usize, usize, usize), WireError> {
- if buf.len() < WIRE_MESSAGE_HEADER_SIZE {
- return Err(WireError::UnexpectedEof {
- offset: 0,
- need: WIRE_MESSAGE_HEADER_SIZE,
- have: buf.len(),
- });
- }
-
- let user_headers_len = u32_at(buf, MSG_USER_HEADERS_LEN_OFFSET) as usize;
- let payload_len = u32_at(buf, MSG_PAYLOAD_LEN_OFFSET) as usize;
-
- let total = WIRE_MESSAGE_HEADER_SIZE
- .checked_add(payload_len)
- .and_then(|s| s.checked_add(user_headers_len))
- .ok_or(WireError::Validation(Cow::Borrowed(
- "message frame size overflow",
- )))?;
-
- if buf.len() < total {
- return Err(WireError::UnexpectedEof {
- offset: 0,
- need: total,
- have: buf.len(),
- });
- }
-
- Ok((total, payload_len, user_headers_len))
-}
-
-// WireMessageView - immutable zero-copy frame view
-
-/// Borrowed view over a single message frame in a contiguous buffer.
-///
-/// Validates the buffer on construction; all accessors are infallible.
-pub struct WireMessageView<'a> {
- buf: &'a [u8],
- payload_len: usize,
- user_headers_len: usize,
-}
-
-impl<'a> WireMessageView<'a> {
- /// Create a view over the first message frame in `buf`.
- ///
- /// # Errors
- /// Returns `WireError` if the buffer is shorter than the header (64 bytes)
- /// or shorter than the full frame indicated by the length fields.
- pub fn new(buf: &'a [u8]) -> Result<Self, WireError> {
- let (total, payload_len, user_headers_len) = validate_frame(buf)?;
- Ok(Self {
- buf: &buf[..total],
- payload_len,
- user_headers_len,
- })
- }
-
- #[must_use]
- #[inline]
- pub fn checksum(&self) -> u64 {
- u64_at(self.buf, MSG_CHECKSUM_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn id(&self) -> u128 {
- u128_at(self.buf, MSG_ID_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn offset(&self) -> u64 {
- u64_at(self.buf, MSG_OFFSET_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn timestamp(&self) -> u64 {
- u64_at(self.buf, MSG_TIMESTAMP_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn origin_timestamp(&self) -> u64 {
- u64_at(self.buf, MSG_ORIGIN_TIMESTAMP_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub const fn payload_length(&self) -> usize {
- self.payload_len
- }
-
- #[must_use]
- #[inline]
- pub const fn user_headers_length(&self) -> usize {
- self.user_headers_len
- }
-
- #[must_use]
- pub fn payload(&self) -> &'a [u8] {
- &self.buf[WIRE_MESSAGE_HEADER_SIZE..WIRE_MESSAGE_HEADER_SIZE + self.payload_len]
- }
-
- #[must_use]
- pub fn user_headers(&self) -> &'a [u8] {
- let start = WIRE_MESSAGE_HEADER_SIZE + self.payload_len;
- &self.buf[start..start + self.user_headers_len]
- }
-
- #[must_use]
- #[inline]
- pub const fn total_size(&self) -> usize {
- self.buf.len()
- }
-
- #[must_use]
- pub const fn as_bytes(&self) -> &'a [u8] {
- self.buf
- }
-}
-
-// WireMessageViewMut - mutable zero-copy frame view
-
-/// Mutable view for in-place header patching (offset, timestamp, checksum).
-pub struct WireMessageViewMut<'a> {
- buf: &'a mut [u8],
- payload_len: usize,
- user_headers_len: usize,
-}
-
-impl<'a> WireMessageViewMut<'a> {
- /// Create a mutable view over the first message frame in `buf`.
- ///
- /// # Errors
- /// Same validation as [`WireMessageView::new`].
- pub fn new(buf: &'a mut [u8]) -> Result<Self, WireError> {
- let (total, payload_len, user_headers_len) = validate_frame(buf)?;
- let (frame, _) = buf.split_at_mut(total);
- Ok(Self {
- buf: frame,
- payload_len,
- user_headers_len,
- })
- }
-
- // -- Read accessors (same semantics as WireMessageView) --
-
- #[must_use]
- #[inline]
- pub fn checksum(&self) -> u64 {
- u64_at(self.buf, MSG_CHECKSUM_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn id(&self) -> u128 {
- u128_at(self.buf, MSG_ID_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn offset(&self) -> u64 {
- u64_at(self.buf, MSG_OFFSET_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn timestamp(&self) -> u64 {
- u64_at(self.buf, MSG_TIMESTAMP_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub fn origin_timestamp(&self) -> u64 {
- u64_at(self.buf, MSG_ORIGIN_TIMESTAMP_OFFSET)
- }
-
- #[must_use]
- #[inline]
- pub const fn payload_length(&self) -> usize {
- self.payload_len
- }
-
- #[must_use]
- #[inline]
- pub const fn user_headers_length(&self) -> usize {
- self.user_headers_len
- }
-
- #[must_use]
- pub fn payload(&self) -> &[u8] {
- &self.buf[WIRE_MESSAGE_HEADER_SIZE..WIRE_MESSAGE_HEADER_SIZE + self.payload_len]
- }
-
- #[must_use]
- pub fn user_headers(&self) -> &[u8] {
- let start = WIRE_MESSAGE_HEADER_SIZE + self.payload_len;
- &self.buf[start..start + self.user_headers_len]
- }
-
- #[must_use]
- #[inline]
- pub const fn total_size(&self) -> usize {
- self.buf.len()
- }
-
- // -- Typed setters --
-
- pub fn set_checksum(&mut self, value: u64) {
- self.buf[MSG_CHECKSUM_OFFSET..MSG_CHECKSUM_OFFSET + 8]
- .copy_from_slice(&value.to_le_bytes());
- }
-
- pub fn set_id(&mut self, value: u128) {
- self.buf[MSG_ID_OFFSET..MSG_ID_OFFSET + 16].copy_from_slice(&value.to_le_bytes());
- }
-
- pub fn set_offset(&mut self, value: u64) {
- self.buf[MSG_OFFSET_OFFSET..MSG_OFFSET_OFFSET + 8].copy_from_slice(&value.to_le_bytes());
- }
-
- pub fn set_timestamp(&mut self, value: u64) {
- self.buf[MSG_TIMESTAMP_OFFSET..MSG_TIMESTAMP_OFFSET + 8]
- .copy_from_slice(&value.to_le_bytes());
- }
-}
-
-// WireMessageIterator - zero-copy frame iterator
-
-/// Iterates over contiguous message frames, yielding borrowed views.
-pub struct WireMessageIterator<'a> {
- buf: &'a [u8],
- pos: usize,
- remaining: u32,
-}
-
-impl<'a> WireMessageIterator<'a> {
- #[must_use]
- pub const fn new(buf: &'a [u8], count: u32) -> Self {
- Self {
- buf,
- pos: 0,
- remaining: count,
- }
- }
-}
-
-impl<'a> Iterator for WireMessageIterator<'a> {
- type Item = Result<WireMessageView<'a>, WireError>;
-
- fn next(&mut self) -> Option<Self::Item> {
- if self.remaining == 0 {
- return None;
- }
- match WireMessageView::new(&self.buf[self.pos..]) {
- Ok(view) => {
- self.pos += view.total_size();
- self.remaining -= 1;
- Some(Ok(view))
- }
- Err(e) => {
- self.remaining = 0;
- Some(Err(e))
- }
- }
- }
-}
-
-// WireMessageIteratorMut - mutable frame iterator (cursor-based)
-
-/// Mutable iterator over contiguous message frames.
-///
-/// Cannot implement standard `Iterator` because the yielded mutable view's
-/// lifetime is tied to `&mut self`. Uses a cursor-based API instead.
-pub struct WireMessageIteratorMut<'a> {
- buf: &'a mut [u8],
- pos: usize,
- remaining: u32,
-}
-
-impl<'a> WireMessageIteratorMut<'a> {
- pub const fn new(buf: &'a mut [u8], count: u32) -> Self {
- Self {
- buf,
- pos: 0,
- remaining: count,
- }
- }
-
- /// Advance to the next frame and return a mutable view over it.
- ///
- /// The returned view borrows `self`, so it must be dropped before
- /// calling this method again.
- pub fn next_view_mut(&mut self) -> Option<Result<WireMessageViewMut<'_>, WireError>> {
- if self.remaining == 0 {
- return None;
- }
-
- // Compute frame size via temporary immutable access (borrow ends at block exit).
- let (total, available) = {
- let rest = &self.buf[self.pos..];
- if rest.len() < WIRE_MESSAGE_HEADER_SIZE {
- self.remaining = 0;
- return Some(Err(WireError::UnexpectedEof {
- offset: self.pos,
- need: WIRE_MESSAGE_HEADER_SIZE,
- have: rest.len(),
- }));
- }
- let user_headers_len = u32_at(rest, MSG_USER_HEADERS_LEN_OFFSET) as usize;
- let payload_len = u32_at(rest, MSG_PAYLOAD_LEN_OFFSET) as usize;
- let Some(total) = WIRE_MESSAGE_HEADER_SIZE
- .checked_add(payload_len)
- .and_then(|s| s.checked_add(user_headers_len))
- else {
- self.remaining = 0;
- return Some(Err(WireError::Validation(Cow::Borrowed(
- "message frame size overflow",
- ))));
- };
- (total, rest.len())
- };
-
- if available < total {
- self.remaining = 0;
- return Some(Err(WireError::UnexpectedEof {
- offset: self.pos,
- need: total,
- have: available,
- }));
- }
-
- let start = self.pos;
- self.pos += total;
- self.remaining -= 1;
-
- Some(WireMessageViewMut::new(&mut self.buf[start..start + total]))
- }
-}
-
-// Tests
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use crate::message_layout::{
- MSG_ID_OFFSET, MSG_ORIGIN_TIMESTAMP_OFFSET, MSG_PAYLOAD_LEN_OFFSET,
- MSG_USER_HEADERS_LEN_OFFSET, WIRE_MESSAGE_HEADER_SIZE,
- };
-
- fn make_frame(payload: &[u8], user_headers: &[u8], id: u128, origin_ts: u64) -> Vec<u8> {
- let total = WIRE_MESSAGE_HEADER_SIZE + payload.len() + user_headers.len();
- let mut frame = vec![0u8; total];
- frame[MSG_ID_OFFSET..MSG_ID_OFFSET + 16].copy_from_slice(&id.to_le_bytes());
- frame[MSG_ORIGIN_TIMESTAMP_OFFSET..MSG_ORIGIN_TIMESTAMP_OFFSET + 8]
- .copy_from_slice(&origin_ts.to_le_bytes());
- #[allow(clippy::cast_possible_truncation)]
- {
- frame[MSG_USER_HEADERS_LEN_OFFSET..MSG_USER_HEADERS_LEN_OFFSET + 4]
- .copy_from_slice(&(user_headers.len() as u32).to_le_bytes());
- frame[MSG_PAYLOAD_LEN_OFFSET..MSG_PAYLOAD_LEN_OFFSET + 4]
- .copy_from_slice(&(payload.len() as u32).to_le_bytes());
- }
- frame[WIRE_MESSAGE_HEADER_SIZE..WIRE_MESSAGE_HEADER_SIZE + payload.len()]
- .copy_from_slice(payload);
- frame[WIRE_MESSAGE_HEADER_SIZE + payload.len()..].copy_from_slice(user_headers);
- frame
- }
-
- // -- WireMessageView --
-
- #[test]
- fn view_rejects_short_buffer() {
- let buf = [0u8; WIRE_MESSAGE_HEADER_SIZE - 1];
- assert!(WireMessageView::new(&buf).is_err());
- }
-
- #[test]
- fn view_rejects_truncated_payload() {
- let mut buf = vec![0u8; WIRE_MESSAGE_HEADER_SIZE + 5];
- buf[MSG_PAYLOAD_LEN_OFFSET..MSG_PAYLOAD_LEN_OFFSET + 4]
- .copy_from_slice(&10u32.to_le_bytes());
- assert!(WireMessageView::new(&buf).is_err());
- }
-
- #[test]
- fn view_accessors() {
- let frame = make_frame(b"hello", b"hdr", 42, 999);
- let view = WireMessageView::new(&frame).unwrap();
- assert_eq!(view.checksum(), 0);
- assert_eq!(view.id(), 42);
- assert_eq!(view.offset(), 0);
- assert_eq!(view.timestamp(), 0);
- assert_eq!(view.origin_timestamp(), 999);
- assert_eq!(view.payload_length(), 5);
- assert_eq!(view.user_headers_length(), 3);
- assert_eq!(view.payload(), b"hello");
- assert_eq!(view.user_headers(), b"hdr");
- assert_eq!(view.total_size(), WIRE_MESSAGE_HEADER_SIZE + 5 + 3);
- assert_eq!(view.as_bytes().len(), view.total_size());
- }
-
- #[test]
- fn view_empty_payload_and_headers() {
- let frame = make_frame(b"", b"", 1, 0);
- let view = WireMessageView::new(&frame).unwrap();
- assert_eq!(view.payload(), b"");
- assert_eq!(view.user_headers(), b"");
- assert_eq!(view.total_size(), WIRE_MESSAGE_HEADER_SIZE);
- }
-
- #[test]
- fn view_from_larger_buffer() {
- let mut buf = make_frame(b"pay", b"", 1, 0);
- buf.extend_from_slice(b"trailing_garbage");
- let view = WireMessageView::new(&buf).unwrap();
- assert_eq!(view.total_size(), WIRE_MESSAGE_HEADER_SIZE + 3);
- assert_eq!(view.as_bytes().len(), WIRE_MESSAGE_HEADER_SIZE + 3);
- }
-
- // -- WireMessageViewMut --
-
- #[test]
- fn view_mut_set_get_roundtrip() {
- let mut frame = make_frame(b"data", b"", 1, 0);
- let mut view = WireMessageViewMut::new(&mut frame).unwrap();
-
- view.set_checksum(0xDEAD_BEEF);
- view.set_id(0x1234);
- view.set_offset(100);
- view.set_timestamp(200);
-
- assert_eq!(view.checksum(), 0xDEAD_BEEF);
- assert_eq!(view.id(), 0x1234);
- assert_eq!(view.offset(), 100);
- assert_eq!(view.timestamp(), 200);
- assert_eq!(view.origin_timestamp(), 0);
- assert_eq!(view.payload(), b"data");
- }
-
- #[test]
- fn view_mut_set_does_not_corrupt_adjacent_fields() {
- let mut frame = make_frame(b"x", b"y", 42, 999);
- {
- let mut view = WireMessageViewMut::new(&mut frame).unwrap();
- view.set_offset(0xFFFF_FFFF_FFFF_FFFF);
- }
- let view = WireMessageView::new(&frame).unwrap();
- assert_eq!(view.id(), 42);
- assert_eq!(view.offset(), 0xFFFF_FFFF_FFFF_FFFF);
- assert_eq!(view.timestamp(), 0);
- assert_eq!(view.origin_timestamp(), 999);
- assert_eq!(view.payload(), b"x");
- assert_eq!(view.user_headers(), b"y");
- }
-
- // -- WireMessageIterator --
-
- #[test]
- fn iterator_zero_count() {
- let buf = [];
- let mut iter = WireMessageIterator::new(&buf, 0);
- assert!(iter.next().is_none());
- }
-
- #[test]
- fn iterator_single_message() {
- let frame = make_frame(b"hello", b"", 1, 100);
- let mut iter = WireMessageIterator::new(&frame, 1);
- let view = iter.next().unwrap().unwrap();
- assert_eq!(view.id(), 1);
- assert_eq!(view.payload(), b"hello");
- assert!(iter.next().is_none());
- }
-
- #[test]
- fn iterator_multiple_messages() {
- let f1 = make_frame(b"aaa", b"", 1, 100);
- let f2 = make_frame(b"bbb", b"hh", 2, 200);
- let f3 = make_frame(b"c", b"", 3, 300);
- let mut buf = Vec::new();
- buf.extend_from_slice(&f1);
- buf.extend_from_slice(&f2);
- buf.extend_from_slice(&f3);
-
- let views: Vec<_> = WireMessageIterator::new(&buf, 3)
- .collect::<Result<Vec<_>, _>>()
- .unwrap();
- assert_eq!(views.len(), 3);
- assert_eq!(views[0].id(), 1);
- assert_eq!(views[0].payload(), b"aaa");
- assert_eq!(views[1].id(), 2);
- assert_eq!(views[1].payload(), b"bbb");
- assert_eq!(views[1].user_headers(), b"hh");
- assert_eq!(views[2].id(), 3);
- assert_eq!(views[2].payload(), b"c");
- }
-
- #[test]
- fn iterator_truncated_mid_frame() {
- let frame = make_frame(b"hello", b"", 1, 0);
- let truncated = &frame[..frame.len() / 2];
- let mut iter = WireMessageIterator::new(truncated, 1);
- assert!(iter.next().unwrap().is_err());
- assert!(iter.next().is_none());
- }
-
- // -- WireMessageIteratorMut --
-
- #[test]
- fn iterator_mut_mutate_then_read() {
- let f1 = make_frame(b"aaa", b"", 1, 100);
- let f2 = make_frame(b"bbb", b"", 2, 200);
- let mut buf = Vec::new();
- buf.extend_from_slice(&f1);
- buf.extend_from_slice(&f2);
-
- let mut iter = WireMessageIteratorMut::new(&mut buf, 2);
- {
- let mut view = iter.next_view_mut().unwrap().unwrap();
- view.set_offset(10);
- view.set_timestamp(1000);
- }
- {
- let mut view = iter.next_view_mut().unwrap().unwrap();
- view.set_offset(20);
- view.set_timestamp(2000);
- }
- assert!(iter.next_view_mut().is_none());
-
- let views: Vec<_> = WireMessageIterator::new(&buf, 2)
- .collect::<Result<Vec<_>, _>>()
- .unwrap();
- assert_eq!(views[0].offset(), 10);
- assert_eq!(views[0].timestamp(), 1000);
- assert_eq!(views[0].payload(), b"aaa");
- assert_eq!(views[1].offset(), 20);
- assert_eq!(views[1].timestamp(), 2000);
- assert_eq!(views[1].payload(), b"bbb");
- }
-
- #[test]
- fn iterator_mut_count_matches() {
- let f1 = make_frame(b"a", b"", 1, 0);
- let f2 = make_frame(b"b", b"", 2, 0);
- let mut buf = Vec::new();
- buf.extend_from_slice(&f1);
- buf.extend_from_slice(&f2);
-
- let mut iter = WireMessageIteratorMut::new(&mut buf, 2);
- let mut count = 0;
- while let Some(Ok(_)) = iter.next_view_mut() {
- count += 1;
- }
- assert_eq!(count, 2);
- }
-}
diff --git a/core/binary_protocol/src/requests/messages/mod.rs b/core/binary_protocol/src/requests/messages/mod.rs
index 6b976e07e..76584099d 100644
--- a/core/binary_protocol/src/requests/messages/mod.rs
+++ b/core/binary_protocol/src/requests/messages/mod.rs
@@ -21,6 +21,4 @@ pub mod send_messages;
pub use flush_unsaved_buffer::FlushUnsavedBufferRequest;
pub use poll_messages::PollMessagesRequest;
-pub use send_messages::{
- RawMessage, SendMessagesEncoder, SendMessagesHeader, SendMessagesMetadataEncoder,
-};
+pub use send_messages::{RawMessage, SendMessagesEncoder, SendMessagesHeader};
diff --git a/core/binary_protocol/src/requests/messages/send_messages.rs b/core/binary_protocol/src/requests/messages/send_messages.rs
index 559ce9e0b..df2bfafa4 100644
--- a/core/binary_protocol/src/requests/messages/send_messages.rs
+++ b/core/binary_protocol/src/requests/messages/send_messages.rs
@@ -15,16 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-//! Zero-copy encoder for `SendMessages` wire format.
+//! Encoder for the `SendMessages` wire format.
//!
-//! Messages are written directly to the buffer without intermediate allocation.
-
+//! The body is the routing metadata section followed by one canonical message
+//! batch (see [`crate::batch`]) - the same record the server replicates,
+//! persists, and serves back to polls.
+
+use crate::batch::{
+ BATCH_HEADER_SIZE, BATCH_MESSAGE_HEADER_SIZE, BatchHeader, MAX_TIMESTAMP_DELTA_MICROS,
+ calculate_batch_checksum,
+};
use crate::codec::{WireDecode, WireEncode, read_u32_le};
use crate::error::WireError;
-use crate::message_layout::{WIRE_MESSAGE_HEADER_SIZE, WIRE_MESSAGE_INDEX_SIZE};
use crate::primitives::identifier::WireIdentifier;
use crate::primitives::partitioning::WirePartitioning;
use bytes::{BufMut, BytesMut};
+use twox_hash::XxHash3_64;
/// Borrowed message data for encoding. No allocation needed - the caller
/// owns the payload and headers buffers.
@@ -36,29 +42,12 @@ pub struct RawMessage<'a> {
}
impl RawMessage<'_> {
- fn wire_size(&self) -> usize {
- WIRE_MESSAGE_HEADER_SIZE + self.payload.len() + self.headers.map_or(0, <[u8]>::len)
- }
-
- /// Write the 64-byte message header to `buf`. Payload and `user_headers` are NOT
- /// written - the caller sends them separately via vectored I/O.
- pub fn encode_header(&self, buf: &mut BytesMut) {
- let headers_len = self.headers.map_or(0, <[u8]>::len);
- buf.put_u64_le(0); // checksum (server-computed)
- buf.put_u128_le(self.id);
- buf.put_u64_le(0); // offset (server-assigned)
- buf.put_u64_le(0); // timestamp (server-assigned)
- buf.put_u64_le(self.origin_timestamp);
- #[allow(clippy::cast_possible_truncation)]
- {
- buf.put_u32_le(headers_len as u32);
- buf.put_u32_le(self.payload.len() as u32);
- }
- buf.put_u64_le(0); // reserved
+ fn frame_size(&self) -> usize {
+ BATCH_MESSAGE_HEADER_SIZE + self.payload.len() + self.headers.map_or(0, <[u8]>::len)
}
}
-/// Zero-copy encoder for the `SendMessages` command payload.
+/// Encoder for the `SendMessages` command payload.
///
/// Wire layout:
/// ```text
@@ -67,9 +56,12 @@ impl RawMessage<'_> {
/// [topic_id:variable]
/// [partitioning:variable]
/// [messages_count:u32_le]
-/// [index_array: messages_count * 16 bytes]
-/// [message_data: variable]
+/// [batch: 256-byte batch header + message frames]
/// ```
+///
+/// The producer leaves `partition_id`, `base_offset`, and `base_timestamp`
+/// zero in the batch header; the server stamps them. Every checksum is
+/// producer-computed and verified at admission.
pub struct SendMessagesEncoder;
impl SendMessagesEncoder {
@@ -82,18 +74,24 @@ impl SendMessagesEncoder {
) -> usize {
let metadata_inner =
stream_id.encoded_size() + topic_id.encoded_size() + partitioning.encoded_size() + 4;
- let index_total = messages.len() * WIRE_MESSAGE_INDEX_SIZE;
- let messages_total: usize = messages.iter().map(RawMessage::wire_size).sum();
- 4 + metadata_inner + index_total + messages_total
+ let blob_total: usize = messages.iter().map(RawMessage::frame_size).sum();
+ 4 + metadata_inner + BATCH_HEADER_SIZE + blob_total
}
+ /// Encode the full `SendMessages` body into `buf`.
+ ///
+ /// # Errors
+ /// [`WireError::InvalidMessageTimestampDelta`] when a message's
+ /// `origin_timestamp` runs more than [`MAX_TIMESTAMP_DELTA_MICROS`] ahead
+ /// of the batch's earliest one; [`WireError::PayloadTooLarge`] when a
+ /// section length or the batch length overflows its wire field.
pub fn encode(
buf: &mut BytesMut,
stream_id: &WireIdentifier,
topic_id: &WireIdentifier,
partitioning: &WirePartitioning,
messages: &[RawMessage<'_>],
- ) {
+ ) -> Result<(), WireError> {
let metadata_inner =
stream_id.encoded_size() + topic_id.encoded_size() + partitioning.encoded_size() + 4;
@@ -104,52 +102,80 @@ impl SendMessagesEncoder {
topic_id.encode(buf);
partitioning.encode(buf);
- #[allow(clippy::cast_possible_truncation)]
- buf.put_u32_le(messages.len() as u32);
-
- // Index array: cumulative sizes for each message
- let mut cumulative_size: u32 = 0;
- for msg in messages {
- #[allow(clippy::cast_possible_truncation)]
- {
- cumulative_size += msg.wire_size() as u32;
+ let message_count =
+ u32::try_from(messages.len()).map_err(|_| WireError::PayloadTooLarge {
+ size: messages.len(),
+ max: u32::MAX as usize,
+ })?;
+ buf.put_u32_le(message_count);
+
+ let origin_timestamp = messages
+ .iter()
+ .map(|message| message.origin_timestamp)
+ .min()
+ .unwrap_or(0);
+
+ // The batch header depends on the frame checksums written below, so
+ // reserve its slot and backpatch once the blob is in place.
+ let header_start = buf.len();
+ buf.resize(header_start + BATCH_HEADER_SIZE, 0);
+
+ let blob_start = buf.len();
+ for (index, message) in messages.iter().enumerate() {
+ let timestamp_delta = message.origin_timestamp - origin_timestamp;
+ if timestamp_delta > MAX_TIMESTAMP_DELTA_MICROS {
+ return Err(WireError::InvalidMessageTimestampDelta(timestamp_delta));
}
- // bytes 0-3: zero
- buf.put_u32_le(0);
- // bytes 4-7: cumulative size
- buf.put_u32_le(cumulative_size);
- // bytes 8-15: zero
- buf.put_u64_le(0);
- }
-
- // Message data: header(64) + payload + optional user_headers
- for msg in messages {
- let headers_len = msg.headers.map_or(0, <[u8]>::len);
-
- buf.put_u64_le(0); // checksum (server-computed)
- buf.put_u128_le(msg.id);
- buf.put_u64_le(0); // offset (server-assigned)
- buf.put_u64_le(0); // timestamp (server-assigned)
- buf.put_u64_le(msg.origin_timestamp);
+ let headers = message.headers.unwrap_or_default();
+ let user_headers_length =
+ u32::try_from(headers.len()).map_err(|_| WireError::PayloadTooLarge {
+ size: headers.len(),
+ max: u32::MAX as usize,
+ })?;
+ let payload_length =
+ u32::try_from(message.payload.len()).map_err(|_| WireError::PayloadTooLarge {
+ size: message.payload.len(),
+ max: u32::MAX as usize,
+ })?;
+
+ let frame_start = buf.len();
+ buf.put_u64_le(0); // checksum, backpatched below
+ buf.put_u128_le(message.id);
#[allow(clippy::cast_possible_truncation)]
{
- buf.put_u32_le(headers_len as u32);
- buf.put_u32_le(msg.payload.len() as u32);
+ buf.put_u32_le(index as u32); // offset_delta
+ buf.put_u32_le(timestamp_delta as u32);
}
+ buf.put_u32_le(user_headers_length);
+ buf.put_u32_le(payload_length);
buf.put_u64_le(0); // reserved
- buf.put_slice(msg.payload);
- if let Some(headers) = msg.headers {
- buf.put_slice(headers);
- }
+ buf.put_slice(message.payload);
+ buf.put_slice(headers);
+
+ let checksum = XxHash3_64::oneshot(&buf[frame_start + 8..]);
+ buf[frame_start..frame_start + 8].copy_from_slice(&checksum.to_le_bytes());
}
+
+ let batch_length =
+ u64::try_from(BATCH_HEADER_SIZE + (buf.len() - blob_start)).map_err(|_| {
+ WireError::PayloadTooLarge {
+ size: buf.len() - blob_start,
+ max: u32::MAX as usize,
+ }
+ })?;
+ let mut batch_header = BatchHeader::new(0, origin_timestamp, batch_length, message_count);
+ batch_header.batch_checksum = calculate_batch_checksum(&batch_header, &buf[blob_start..]);
+ let header_bytes: &mut [u8] = &mut buf[header_start..header_start + BATCH_HEADER_SIZE];
+ batch_header.encode_into(header_bytes);
+ Ok(())
}
}
/// Metadata-only decoder for the `SendMessages` command.
///
/// Parses routing metadata (stream, topic, partitioning, message count)
-/// without touching message payloads. The server uses this to extract
-/// routing info before reading message data into pooled buffers.
+/// without touching the batch. The server uses this to resolve the namespace
+/// before validating the batch bytes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SendMessagesHeader {
pub stream_id: WireIdentifier,
@@ -204,145 +230,75 @@ impl WireDecode for SendMessagesHeader {
}
}
-/// Vectored I/O encoder for `SendMessages`.
-///
-/// Writes metadata + index array but stops before message frame data,
-/// enabling the caller to compose `[header_buf | msg_frames...]` via writev.
-pub struct SendMessagesMetadataEncoder;
-
-impl SendMessagesMetadataEncoder {
- /// Exact size of metadata + indexes (everything except message frame data).
- #[must_use]
- pub fn header_size(
- stream_id: &WireIdentifier,
- topic_id: &WireIdentifier,
- partitioning: &WirePartitioning,
- messages_count: usize,
- ) -> usize {
- let metadata_inner =
- stream_id.encoded_size() + topic_id.encoded_size() + partitioning.encoded_size() + 4;
- 4 + metadata_inner + messages_count * WIRE_MESSAGE_INDEX_SIZE
- }
-
- /// Encode `metadata_length` + metadata fields + index array into `buf`.
- ///
- /// Does NOT write message frame data - the caller sends that via vectored I/O.
- pub fn encode_header(
- buf: &mut BytesMut,
- stream_id: &WireIdentifier,
- topic_id: &WireIdentifier,
- partitioning: &WirePartitioning,
- messages: &[RawMessage<'_>],
- ) {
- let metadata_inner =
- stream_id.encoded_size() + topic_id.encoded_size() + partitioning.encoded_size() + 4;
-
- #[allow(clippy::cast_possible_truncation)]
- buf.put_u32_le(metadata_inner as u32);
-
- stream_id.encode(buf);
- topic_id.encode(buf);
- partitioning.encode(buf);
-
- #[allow(clippy::cast_possible_truncation)]
- buf.put_u32_le(messages.len() as u32);
-
- let mut cumulative_size: u32 = 0;
- for msg in messages {
- #[allow(clippy::cast_possible_truncation)]
- {
- cumulative_size += msg.wire_size() as u32;
- }
- buf.put_u32_le(0);
- buf.put_u32_le(cumulative_size);
- buf.put_u64_le(0);
- }
- }
-}
-
#[cfg(test)]
mod tests {
use super::*;
+ use crate::batch::decode_batch_slice;
fn numeric_id(id: u32) -> WireIdentifier {
WireIdentifier::numeric(id)
}
- #[test]
- fn encode_single_message_no_headers() {
- let stream_id = numeric_id(1);
- let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::Balanced;
- let payload = b"hello";
- let messages = [RawMessage {
- id: 100,
- origin_timestamp: 999,
- headers: None,
- payload: payload.as_slice(),
- }];
-
- let size =
- SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
+ fn encode(
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ partitioning: &WirePartitioning,
+ messages: &[RawMessage<'_>],
+ ) -> BytesMut {
+ let size = SendMessagesEncoder::encoded_size(stream_id, topic_id, partitioning, messages);
let mut buf = BytesMut::with_capacity(size);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
-
+ SendMessagesEncoder::encode(&mut buf, stream_id, topic_id, partitioning, messages)
+ .expect("encodes");
assert_eq!(buf.len(), size);
+ buf
}
#[test]
- fn encode_single_message_with_headers() {
+ fn encoded_batch_decodes_and_verifies() {
let stream_id = numeric_id(1);
let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::PartitionId(5);
- let payload = b"world";
- let headers = b"key:val";
- let messages = [RawMessage {
- id: 200,
- origin_timestamp: 1000,
- headers: Some(headers.as_slice()),
- payload: payload.as_slice(),
- }];
-
- let size =
- SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
- let mut buf = BytesMut::with_capacity(size);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
-
- assert_eq!(buf.len(), size);
- }
-
- #[test]
- fn encode_multiple_messages() {
- let stream_id = numeric_id(10);
- let topic_id = numeric_id(20);
- let partitioning = WirePartitioning::MessagesKey(b"user-1".to_vec());
+ let partitioning = WirePartitioning::Balanced;
let messages = [
RawMessage {
- id: 1,
- origin_timestamp: 100,
+ id: 100,
+ origin_timestamp: 1_000,
headers: None,
- payload: b"msg-1",
- },
- RawMessage {
- id: 2,
- origin_timestamp: 200,
- headers: Some(b"h2"),
- payload: b"msg-2",
+ payload: b"first",
},
RawMessage {
- id: 3,
- origin_timestamp: 300,
- headers: None,
- payload: b"msg-3",
+ id: 200,
+ origin_timestamp: 2_000,
+ headers: Some(b"hdr"),
+ payload: b"second",
},
];
- let size =
- SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
- let mut buf = BytesMut::with_capacity(size);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
+ let buf = encode(&stream_id, &topic_id, &partitioning, &messages);
- assert_eq!(buf.len(), size);
+ let metadata_len = u32::from_le_bytes(buf[0..4].try_into().unwrap()) as usize;
+ let (header, consumed) = SendMessagesHeader::decode(&buf[4..4 + metadata_len]).unwrap();
+ assert_eq!(consumed, metadata_len);
+ assert_eq!(header.messages_count, 2);
+
+ let batch = decode_batch_slice(&buf[4 + metadata_len..]).unwrap();
+ assert_eq!(batch.header.partition_id, 0);
+ assert_eq!(batch.header.base_offset, 0);
+ assert_eq!(batch.header.base_timestamp, 0);
+ assert_eq!(batch.header.origin_timestamp, 1_000);
+ assert_eq!(batch.header.total_size(), buf.len() - 4 - metadata_len);
+ assert_eq!(batch.message_count(), 2);
+
+ let views: Vec<_> = batch.iter().collect();
+ assert_eq!(views[0].header.id, 100);
+ assert_eq!(views[0].header.offset_delta, 0);
+ assert_eq!(views[0].header.timestamp_delta, 0);
+ assert_eq!(views[0].payload, b"first");
+ assert_eq!(views[0].user_headers, b"");
+ assert_eq!(views[1].header.id, 200);
+ assert_eq!(views[1].header.offset_delta, 1);
+ assert_eq!(views[1].header.timestamp_delta, 1_000);
+ assert_eq!(views[1].payload, b"second");
+ assert_eq!(views[1].user_headers, b"hdr");
}
#[test]
@@ -357,8 +313,7 @@ mod tests {
payload: b"x",
}];
- let mut buf = BytesMut::with_capacity(256);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
+ let buf = encode(&stream_id, &topic_id, &partitioning, &messages);
let metadata_len = u32::from_le_bytes(buf[0..4].try_into().unwrap()) as usize;
let expected =
@@ -366,158 +321,51 @@ mod tests {
assert_eq!(metadata_len, expected);
}
- #[test]
- fn verify_index_entries() {
- let stream_id = numeric_id(1);
- let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::Balanced;
- let messages = [
- RawMessage {
- id: 1,
- origin_timestamp: 0,
- headers: None,
- payload: b"aaaa", // wire_size = 64 + 4 = 68
- },
- RawMessage {
- id: 2,
- origin_timestamp: 0,
- headers: Some(b"hh"), // wire_size = 64 + 3 + 2 = 69
- payload: b"bbb",
- },
- ];
-
- let mut buf = BytesMut::with_capacity(512);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
-
- // Index starts after: 4 (metadata_len) + stream_id(6) + topic_id(6) + partitioning(2) + 4 (msg_count)
- let index_start = 4 + 6 + 6 + 2 + 4;
-
- // First index entry: cumulative = 68
- let first_zero = u32::from_le_bytes(buf[index_start..index_start + 4].try_into().unwrap());
- let first_cum =
- u32::from_le_bytes(buf[index_start + 4..index_start + 8].try_into().unwrap());
- assert_eq!(first_zero, 0);
- assert_eq!(first_cum, 68);
-
- // Second index entry: cumulative = 68 + 69 = 137
- let second_start = index_start + WIRE_MESSAGE_INDEX_SIZE;
- let second_zero =
- u32::from_le_bytes(buf[second_start..second_start + 4].try_into().unwrap());
- let second_cum =
- u32::from_le_bytes(buf[second_start + 4..second_start + 8].try_into().unwrap());
- assert_eq!(second_zero, 0);
- assert_eq!(second_cum, 137);
- }
-
- #[test]
- fn verify_message_header_layout() {
- let stream_id = numeric_id(1);
- let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::Balanced;
- let messages = [RawMessage {
- id: 42,
- origin_timestamp: 777,
- headers: Some(b"hdr"),
- payload: b"pay",
- }];
-
- let mut buf = BytesMut::with_capacity(256);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
-
- // Message data starts after: 4 + 6 + 6 + 2 + 4 + 16 (one index entry)
- let msg_start = 4 + 6 + 6 + 2 + 4 + 16;
- let msg = &buf[msg_start..];
-
- // Header layout matches WIRE_MESSAGE_HEADER_SIZE (64 bytes):
- // checksum(8) + id(16) + offset(8) + timestamp(8) + origin_ts(8) + hdrs_len(4) + payload_len(4) + reserved(8)
- let checksum = u64::from_le_bytes(msg[0..8].try_into().unwrap());
- assert_eq!(checksum, 0);
-
- let id = u128::from_le_bytes(msg[8..24].try_into().unwrap());
- assert_eq!(id, 42);
-
- let offset = u64::from_le_bytes(msg[24..32].try_into().unwrap());
- assert_eq!(offset, 0);
-
- let timestamp = u64::from_le_bytes(msg[32..40].try_into().unwrap());
- assert_eq!(timestamp, 0);
-
- let origin_ts = u64::from_le_bytes(msg[40..48].try_into().unwrap());
- assert_eq!(origin_ts, 777);
-
- let headers_len = u32::from_le_bytes(msg[48..52].try_into().unwrap());
- assert_eq!(headers_len, 3);
-
- let payload_len = u32::from_le_bytes(msg[52..56].try_into().unwrap());
- assert_eq!(payload_len, 3);
-
- let reserved = u64::from_le_bytes(msg[56..64].try_into().unwrap());
- assert_eq!(reserved, 0);
-
- // After the 64-byte header: payload then headers
- assert_eq!(&msg[64..67], b"pay");
- assert_eq!(&msg[67..70], b"hdr");
- }
-
#[test]
fn empty_payload_message() {
let stream_id = numeric_id(1);
let topic_id = numeric_id(1);
let partitioning = WirePartitioning::Balanced;
let messages = [RawMessage {
- id: 0,
+ id: 7,
origin_timestamp: 0,
headers: None,
payload: b"",
}];
- let size =
- SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
- let mut buf = BytesMut::with_capacity(size);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
- assert_eq!(buf.len(), size);
+ let buf = encode(&stream_id, &topic_id, &partitioning, &messages);
+ let metadata_len = u32::from_le_bytes(buf[0..4].try_into().unwrap()) as usize;
+ let batch = decode_batch_slice(&buf[4 + metadata_len..]).unwrap();
+ assert_eq!(batch.message_count(), 1);
}
- // -- RawMessage::encode_header --
-
#[test]
- fn raw_message_encode_header_is_64_bytes() {
- let msg = RawMessage {
- id: 42,
- origin_timestamp: 999,
- headers: Some(b"hdr"),
- payload: b"pay",
- };
- let mut buf = BytesMut::with_capacity(WIRE_MESSAGE_HEADER_SIZE);
- msg.encode_header(&mut buf);
- assert_eq!(buf.len(), WIRE_MESSAGE_HEADER_SIZE);
- }
+ fn timestamp_delta_overflow_rejected() {
+ let stream_id = numeric_id(1);
+ let topic_id = numeric_id(1);
+ let partitioning = WirePartitioning::Balanced;
+ let messages = [
+ RawMessage {
+ id: 1,
+ origin_timestamp: 0,
+ headers: None,
+ payload: b"a",
+ },
+ RawMessage {
+ id: 2,
+ origin_timestamp: MAX_TIMESTAMP_DELTA_MICROS + 1,
+ headers: None,
+ payload: b"b",
+ },
+ ];
- #[test]
- fn raw_message_encode_header_field_values() {
- let msg = RawMessage {
- id: 0x1234_5678_9ABC_DEF0_1234_5678_9ABC_DEF0,
- origin_timestamp: 0xCAFE_BABE,
- headers: Some(b"hdr"),
- payload: b"pay",
- };
- let mut buf = BytesMut::with_capacity(WIRE_MESSAGE_HEADER_SIZE);
- msg.encode_header(&mut buf);
-
- assert_eq!(u64::from_le_bytes(buf[0..8].try_into().unwrap()), 0);
- assert_eq!(
- u128::from_le_bytes(buf[8..24].try_into().unwrap()),
- 0x1234_5678_9ABC_DEF0_1234_5678_9ABC_DEF0
- );
- assert_eq!(u64::from_le_bytes(buf[24..32].try_into().unwrap()), 0);
- assert_eq!(u64::from_le_bytes(buf[32..40].try_into().unwrap()), 0);
- assert_eq!(
- u64::from_le_bytes(buf[40..48].try_into().unwrap()),
- 0xCAFE_BABE
- );
- assert_eq!(u32::from_le_bytes(buf[48..52].try_into().unwrap()), 3);
- assert_eq!(u32::from_le_bytes(buf[52..56].try_into().unwrap()), 3);
- assert_eq!(u64::from_le_bytes(buf[56..64].try_into().unwrap()), 0);
+ let mut buf = BytesMut::new();
+ let result =
+ SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
+ assert!(matches!(
+ result,
+ Err(WireError::InvalidMessageTimestampDelta(_))
+ ));
}
// -- SendMessagesHeader --
@@ -580,171 +428,4 @@ mod tests {
assert_eq!(header.metadata_length(), 18);
assert_eq!(header.encoded_size(), 18);
}
-
- #[test]
- fn send_messages_header_cross_validate_with_encoder() {
- let stream_id = numeric_id(1);
- let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::Balanced;
- let messages = [RawMessage {
- id: 42,
- origin_timestamp: 777,
- headers: Some(b"hdr"),
- payload: b"pay",
- }];
-
- let mut buf = BytesMut::with_capacity(256);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
-
- let metadata_len = u32::from_le_bytes(buf[0..4].try_into().unwrap()) as usize;
- let (header, consumed) = SendMessagesHeader::decode(&buf[4..4 + metadata_len]).unwrap();
- assert_eq!(consumed, metadata_len);
- assert_eq!(header.stream_id, stream_id);
- assert_eq!(header.topic_id, topic_id);
- assert_eq!(header.partitioning, partitioning);
- assert_eq!(header.messages_count, 1);
- }
-
- // -- SendMessagesMetadataEncoder --
-
- #[test]
- fn metadata_encoder_header_size_matches_encoded() {
- let stream_id = numeric_id(1);
- let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::Balanced;
- let messages = [
- RawMessage {
- id: 1,
- origin_timestamp: 100,
- headers: None,
- payload: b"msg-1",
- },
- RawMessage {
- id: 2,
- origin_timestamp: 200,
- headers: Some(b"h"),
- payload: b"msg-2",
- },
- ];
-
- let expected_size = SendMessagesMetadataEncoder::header_size(
- &stream_id,
- &topic_id,
- &partitioning,
- messages.len(),
- );
- let mut buf = BytesMut::with_capacity(expected_size);
- SendMessagesMetadataEncoder::encode_header(
- &mut buf,
- &stream_id,
- &topic_id,
- &partitioning,
- &messages,
- );
- assert_eq!(buf.len(), expected_size);
- }
-
- #[test]
- fn metadata_encoder_concat_matches_full_encoder() {
- let stream_id = numeric_id(1);
- let topic_id = numeric_id(2);
- let partitioning = WirePartitioning::Balanced;
- let messages = [
- RawMessage {
- id: 1,
- origin_timestamp: 100,
- headers: None,
- payload: b"aaa",
- },
- RawMessage {
- id: 2,
- origin_timestamp: 200,
- headers: Some(b"hh"),
- payload: b"bbb",
- },
- ];
-
- let full_size =
- SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
- let mut full_buf = BytesMut::with_capacity(full_size);
- SendMessagesEncoder::encode(
- &mut full_buf,
- &stream_id,
- &topic_id,
- &partitioning,
- &messages,
- );
-
- let mut vec_buf = BytesMut::with_capacity(full_size);
- SendMessagesMetadataEncoder::encode_header(
- &mut vec_buf,
- &stream_id,
- &topic_id,
- &partitioning,
- &messages,
- );
- for msg in &messages {
- msg.encode_header(&mut vec_buf);
- vec_buf.put_slice(msg.payload);
- if let Some(headers) = msg.headers {
- vec_buf.put_slice(headers);
- }
- }
-
- assert_eq!(vec_buf.len(), full_buf.len());
- assert_eq!(&vec_buf[..], &full_buf[..]);
- }
-
- // -- Cross-validation: encoder -> header decoder -> iterator --
-
- #[test]
- fn cross_validation_encoder_to_iterator() {
- use crate::message_view::WireMessageIterator;
-
- let stream_id = WireIdentifier::numeric(1);
- let topic_id = WireIdentifier::numeric(2);
- let partitioning = WirePartitioning::Balanced;
- let messages = [
- RawMessage {
- id: 100,
- origin_timestamp: 1000,
- headers: None,
- payload: b"first",
- },
- RawMessage {
- id: 200,
- origin_timestamp: 2000,
- headers: Some(b"hdr"),
- payload: b"second",
- },
- ];
-
- let size =
- SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages);
- let mut buf = BytesMut::with_capacity(size);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages);
-
- let metadata_len = u32::from_le_bytes(buf[0..4].try_into().unwrap()) as usize;
- let (header, _) = SendMessagesHeader::decode(&buf[4..4 + metadata_len]).unwrap();
- assert_eq!(header.messages_count, 2);
-
- let data_offset =
- 4 + metadata_len + (header.messages_count as usize) * WIRE_MESSAGE_INDEX_SIZE;
- let message_data = &buf[data_offset..];
-
- let views: Vec<_> = WireMessageIterator::new(message_data, header.messages_count)
- .collect::<Result<Vec<_>, _>>()
- .unwrap();
- assert_eq!(views.len(), 2);
-
- assert_eq!(views[0].id(), 100);
- assert_eq!(views[0].origin_timestamp(), 1000);
- assert_eq!(views[0].payload(), b"first");
- assert_eq!(views[0].user_headers(), b"");
-
- assert_eq!(views[1].id(), 200);
- assert_eq!(views[1].origin_timestamp(), 2000);
- assert_eq!(views[1].payload(), b"second");
- assert_eq!(views[1].user_headers(), b"hdr");
- }
}
diff --git a/core/binary_protocol/src/responses/messages/poll_messages.rs b/core/binary_protocol/src/responses/messages/poll_messages.rs
index cf31e4475..b2949f2e7 100644
--- a/core/binary_protocol/src/responses/messages/poll_messages.rs
+++ b/core/binary_protocol/src/responses/messages/poll_messages.rs
@@ -15,9 +15,18 @@
// specific language governing permissions and limitations
// under the License.
+//! Decoder for the `PollMessages` response.
+//!
+//! The body after the 16-byte prefix is a stream of canonical batch records
+//! (see [`crate::batch`]) served as stored: each record's header carries the
+//! stamped `base_offset` / `base_timestamp`, and each frame's deltas resolve
+//! against them. A record may be a server-sliced view of a larger stored
+//! batch, so `base_offset + offset_delta` of the first frame is the first
+//! polled offset, not necessarily `base_offset` itself.
+
+use crate::batch::{BATCH_HEADER_SIZE, BatchHeader, BatchIterator};
use crate::codec::{WireDecode, WireEncode, read_u32_le, read_u64_le};
use crate::error::WireError;
-use crate::message_view::WireMessageIterator;
use bytes::{BufMut, BytesMut};
/// Size of the `PollMessages` response header: `partition_id(4) + current_offset(8) + count(4)`.
@@ -61,27 +70,122 @@ impl WireDecode for PollMessagesResponseHeader {
}
}
+/// One polled message with its deltas resolved to absolute values.
+#[derive(Debug, Clone, Copy)]
+pub struct PolledMessageView<'a> {
+ /// Stored per-message checksum, passed through as served.
+ pub checksum: u64,
+ pub id: u128,
+ pub offset: u64,
+ /// Broker append time: the flat batch `base_timestamp` (the per-message
+ /// delta applies to `origin_timestamp` only).
+ pub timestamp: u64,
+ pub origin_timestamp: u64,
+ pub payload: &'a [u8],
+ pub user_headers: &'a [u8],
+}
+
+/// Iterator over every message in a stream of served batch records.
+///
+/// Walks records by `batch_length` and flattens their frames. Framing errors
+/// surface as `Some(Err(_))` and end the iteration.
+pub struct PolledBatchesIterator<'a> {
+ batches: &'a [u8],
+ position: usize,
+ current_header: Option<BatchHeader>,
+ frames: Option<BatchIterator<'a>>,
+ failed: bool,
+}
+
+impl<'a> PolledBatchesIterator<'a> {
+ #[must_use]
+ pub const fn new(batches: &'a [u8]) -> Self {
+ Self {
+ batches,
+ position: 0,
+ current_header: None,
+ frames: None,
+ failed: false,
+ }
+ }
+
+ fn advance_batch(&mut self) -> Result<bool, WireError> {
+ if self.position >= self.batches.len() {
+ return Ok(false);
+ }
+ let header = BatchHeader::decode(&self.batches[self.position..])?;
+ let blob_len = header.blob_len()?;
+ let blob_start = self.position + BATCH_HEADER_SIZE;
+ let blob = self
+ .batches
+ .get(blob_start..blob_start + blob_len)
+ .ok_or_else(|| WireError::UnexpectedEof {
+ offset: self.position,
+ need: header.total_size(),
+ have: self.batches.len() - self.position,
+ })?;
+ self.position += header.total_size();
+ self.frames = Some(crate::batch::BatchRef::new(header, blob).iter());
+ self.current_header = Some(header);
+ Ok(true)
+ }
+}
+
+impl<'a> Iterator for PolledBatchesIterator<'a> {
+ type Item = Result<PolledMessageView<'a>, WireError>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ if self.failed {
+ return None;
+ }
+ loop {
+ if let Some(frames) = &mut self.frames
+ && let Some(view) = frames.next()
+ {
+ let header = self.current_header.expect("frames imply a current header");
+ return Some(Ok(PolledMessageView {
+ checksum: view.header.checksum,
+ id: view.header.id,
+ offset: header.base_offset + u64::from(view.header.offset_delta),
+ timestamp: header.base_timestamp,
+ origin_timestamp: header.origin_timestamp
+ + u64::from(view.header.timestamp_delta),
+ payload: view.payload,
+ user_headers: view.user_headers,
+ }));
+ }
+ match self.advance_batch() {
+ Ok(true) => {}
+ Ok(false) => return None,
+ Err(error) => {
+ self.failed = true;
+ return Some(Err(error));
+ }
+ }
+ }
+ }
+}
+
/// Borrowed `PollMessages` response. Does not own message data.
///
/// Does NOT implement `WireDecode` (trait returns owned data, we borrow).
/// Use [`PollMessagesResponse::decode`] instead.
pub struct PollMessagesResponse<'a> {
pub header: PollMessagesResponseHeader,
- pub messages: WireMessageIterator<'a>,
+ pub messages: PolledBatchesIterator<'a>,
}
impl<'a> PollMessagesResponse<'a> {
/// Decode from a response payload buffer. Borrows the buffer.
///
- /// Reads the 16-byte header then creates an iterator over the remaining
- /// message frames. Messages are validated lazily during iteration.
+ /// Reads the 16-byte header then creates an iterator over the batch
+ /// records that follow. Records are validated lazily during iteration.
///
/// # Errors
/// Returns `WireError` if the buffer is too short for the response header.
pub fn decode(buf: &'a [u8]) -> Result<Self, WireError> {
let (header, _) = PollMessagesResponseHeader::decode(buf)?;
- let messages =
- WireMessageIterator::new(&buf[POLL_RESPONSE_HEADER_SIZE..], header.messages_count);
+ let messages = PolledBatchesIterator::new(&buf[POLL_RESPONSE_HEADER_SIZE..]);
Ok(Self { header, messages })
}
}
@@ -89,88 +193,140 @@ impl<'a> PollMessagesResponse<'a> {
#[cfg(test)]
mod tests {
use super::*;
- use crate::message_layout::{
- MSG_ID_OFFSET, MSG_ORIGIN_TIMESTAMP_OFFSET, MSG_PAYLOAD_LEN_OFFSET,
- WIRE_MESSAGE_HEADER_SIZE,
- };
-
- fn make_frame(payload: &[u8], id: u128, origin_ts: u64) -> Vec<u8> {
- let total = WIRE_MESSAGE_HEADER_SIZE + payload.len();
- let mut frame = vec![0u8; total];
- frame[MSG_ID_OFFSET..MSG_ID_OFFSET + 16].copy_from_slice(&id.to_le_bytes());
- frame[MSG_ORIGIN_TIMESTAMP_OFFSET..MSG_ORIGIN_TIMESTAMP_OFFSET + 8]
- .copy_from_slice(&origin_ts.to_le_bytes());
- #[allow(clippy::cast_possible_truncation)]
- frame[MSG_PAYLOAD_LEN_OFFSET..MSG_PAYLOAD_LEN_OFFSET + 4]
- .copy_from_slice(&(payload.len() as u32).to_le_bytes());
- frame[WIRE_MESSAGE_HEADER_SIZE..].copy_from_slice(payload);
- frame
+ use crate::batch::{BATCH_MESSAGE_HEADER_SIZE, calculate_batch_checksum};
+ use twox_hash::XxHash3_64;
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn frame(id: u128, offset_delta: u32, timestamp_delta: u32, payload: &[u8]) -> Vec<u8> {
+ let mut bytes = vec![0u8; BATCH_MESSAGE_HEADER_SIZE];
+ bytes[8..24].copy_from_slice(&id.to_le_bytes());
+ bytes[24..28].copy_from_slice(&offset_delta.to_le_bytes());
+ bytes[28..32].copy_from_slice(×tamp_delta.to_le_bytes());
+ bytes[36..40].copy_from_slice(&(payload.len() as u32).to_le_bytes());
+ bytes.extend_from_slice(payload);
+ let checksum = XxHash3_64::oneshot(&bytes[8..]);
+ bytes[0..8].copy_from_slice(&checksum.to_le_bytes());
+ bytes
+ }
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn batch_record(
+ base_offset: u64,
+ base_timestamp: u64,
+ origin_timestamp: u64,
+ frames: &[Vec<u8>],
+ ) -> Vec<u8> {
+ let blob: Vec<u8> = frames.concat();
+ let mut header = BatchHeader::new(
+ 1,
+ origin_timestamp,
+ (BATCH_HEADER_SIZE + blob.len()) as u64,
+ frames.len() as u32,
+ );
+ header.base_offset = base_offset;
+ header.base_timestamp = base_timestamp;
+ header.batch_checksum = calculate_batch_checksum(&header, &blob);
+ let mut bytes = vec![0u8; BATCH_HEADER_SIZE];
+ header.encode_into(&mut bytes);
+ bytes.extend_from_slice(&blob);
+ bytes
+ }
+
+ fn response_body(current_offset: u64, count: u32, batches: &[Vec<u8>]) -> Vec<u8> {
+ let mut body = Vec::new();
+ body.extend_from_slice(&7u32.to_le_bytes());
+ body.extend_from_slice(¤t_offset.to_le_bytes());
+ body.extend_from_slice(&count.to_le_bytes());
+ for batch in batches {
+ body.extend_from_slice(batch);
+ }
+ body
}
#[test]
fn response_header_roundtrip() {
let header = PollMessagesResponseHeader {
- partition_id: 7,
- current_offset: 12345,
- messages_count: 3,
+ partition_id: 3,
+ current_offset: 42,
+ messages_count: 7,
};
let bytes = header.to_bytes();
- assert_eq!(bytes.len(), POLL_RESPONSE_HEADER_SIZE);
let (decoded, consumed) = PollMessagesResponseHeader::decode(&bytes).unwrap();
- assert_eq!(consumed, POLL_RESPONSE_HEADER_SIZE);
+ assert_eq!(consumed, bytes.len());
assert_eq!(decoded, header);
}
#[test]
fn response_header_truncation() {
- let header = PollMessagesResponseHeader {
- partition_id: 1,
- current_offset: 0,
- messages_count: 0,
- };
- let bytes = header.to_bytes();
- for i in 0..bytes.len() {
- assert!(
- PollMessagesResponseHeader::decode(&bytes[..i]).is_err(),
- "expected error for truncation at byte {i}"
- );
- }
+ let bytes = [0u8; POLL_RESPONSE_HEADER_SIZE - 1];
+ assert!(PollMessagesResponseHeader::decode(&bytes).is_err());
}
#[test]
- fn response_decode_with_messages() {
- let frame1 = make_frame(b"hello", 1, 100);
- let frame2 = make_frame(b"world", 2, 200);
-
- let mut buf = BytesMut::new();
- buf.put_u32_le(42);
- buf.put_u64_le(999);
- buf.put_u32_le(2);
- buf.put_slice(&frame1);
- buf.put_slice(&frame2);
-
- let resp = PollMessagesResponse::decode(&buf).unwrap();
- assert_eq!(resp.header.partition_id, 42);
- assert_eq!(resp.header.current_offset, 999);
- assert_eq!(resp.header.messages_count, 2);
-
- let views: Vec<_> = resp.messages.collect::<Result<Vec<_>, _>>().unwrap();
- assert_eq!(views.len(), 2);
- assert_eq!(views[0].id(), 1);
- assert_eq!(views[0].payload(), b"hello");
- assert_eq!(views[1].id(), 2);
- assert_eq!(views[1].payload(), b"world");
+ fn decodes_messages_across_batches() {
+ let first = batch_record(
+ 100,
+ 5_000,
+ 1_000,
+ &[frame(11, 0, 0, b"a"), frame(12, 1, 10, b"b")],
+ );
+ let second = batch_record(102, 6_000, 2_000, &[frame(13, 0, 0, b"c")]);
+ let body = response_body(102, 3, &[first, second]);
+
+ let response = PollMessagesResponse::decode(&body).unwrap();
+ assert_eq!(response.header.partition_id, 7);
+ assert_eq!(response.header.current_offset, 102);
+ assert_eq!(response.header.messages_count, 3);
+
+ let messages: Vec<_> = response.messages.collect::<Result<Vec<_>, _>>().unwrap();
+ assert_eq!(messages.len(), 3);
+
+ assert_eq!(messages[0].id, 11);
+ assert_eq!(messages[0].offset, 100);
+ assert_eq!(messages[0].timestamp, 5_000);
+ assert_eq!(messages[0].origin_timestamp, 1_000);
+ assert_eq!(messages[0].payload, b"a");
+
+ assert_eq!(messages[1].id, 12);
+ assert_eq!(messages[1].offset, 101);
+ assert_eq!(messages[1].timestamp, 5_000);
+ assert_eq!(messages[1].origin_timestamp, 1_010);
+ assert_eq!(messages[1].payload, b"b");
+
+ assert_eq!(messages[2].id, 13);
+ assert_eq!(messages[2].offset, 102);
+ assert_eq!(messages[2].timestamp, 6_000);
+ assert_eq!(messages[2].payload, b"c");
+ }
+
+ #[test]
+ fn sliced_record_resolves_leading_delta() {
+ // A server-sliced record keeps the stored base_offset; the first
+ // frame's delta positions it inside the original batch.
+ let record = batch_record(50, 9_000, 0, &[frame(1, 3, 0, b"tail")]);
+ let body = response_body(53, 1, &[record]);
+
+ let response = PollMessagesResponse::decode(&body).unwrap();
+ let messages: Vec<_> = response.messages.collect::<Result<Vec<_>, _>>().unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].offset, 53);
}
#[test]
- fn response_decode_zero_messages() {
- let mut buf = BytesMut::new();
- buf.put_u32_le(1);
- buf.put_u64_le(0);
- buf.put_u32_le(0);
-
- let resp = PollMessagesResponse::decode(&buf).unwrap();
- assert_eq!(resp.header.messages_count, 0);
- assert_eq!(resp.messages.count(), 0);
+ fn empty_body_yields_no_messages() {
+ let body = response_body(0, 0, &[]);
+ let response = PollMessagesResponse::decode(&body).unwrap();
+ assert_eq!(response.messages.count(), 0);
+ }
+
+ #[test]
+ fn truncated_record_surfaces_error() {
+ let record = batch_record(0, 0, 0, &[frame(1, 0, 0, b"payload")]);
+ let mut body = response_body(0, 1, &[record]);
+ body.truncate(body.len() - 1);
+
+ let response = PollMessagesResponse::decode(&body).unwrap();
+ let result: Result<Vec<_>, _> = response.messages.collect();
+ assert!(result.is_err());
}
}
diff --git a/core/common/src/traits/binary_impls/messages.rs b/core/common/src/traits/binary_impls/messages.rs
index ab1f26004..bf720a1fe 100644
--- a/core/common/src/traits/binary_impls/messages.rs
+++ b/core/common/src/traits/binary_impls/messages.rs
@@ -337,6 +337,14 @@ impl<B: BinaryClient> MessageClient for B {
let wire_stream_id = identifier_to_wire(stream_id)?;
let wire_topic_id = identifier_to_wire(topic_id)?;
let wire_partitioning = partitioning_to_wire(partitioning)?;
+ // The producer owns message ids now that batches ride the wire
+ // verbatim: a zero id is minted here, before the frame checksum
+ // covers it.
+ for message in messages.iter_mut() {
+ if message.header.id == 0 {
+ message.header.id = crate::utils::random_id::get_uuid();
+ }
+ }
let raw_messages: Vec<RawMessage<'_>> = messages
.iter()
.map(|m| RawMessage {
@@ -359,7 +367,13 @@ impl<B: BinaryClient> MessageClient for B {
&wire_topic_id,
&wire_partitioning,
&raw_messages,
- );
+ )
+ .map_err(|error| match error {
+ iggy_binary_protocol::WireError::InvalidMessageTimestampDelta(delta) => {
+ IggyError::InvalidMessageTimestampDelta(delta)
+ }
+ _ => IggyError::InvalidCommand,
+ })?;
let response = self
.send_raw_with_response(SEND_MESSAGES_CODE, buf.freeze())
.await?;
diff --git a/core/common/src/types/message/message_view_mut.rs b/core/common/src/types/message/message_view_mut.rs
deleted file mode 100644
index 574eb58a2..000000000
--- a/core/common/src/types/message/message_view_mut.rs
+++ /dev/null
@@ -1,117 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use crate::{
- IGGY_MESSAGE_HEADER_SIZE, IggyMessageHeaderView, calculate_checksum,
- types::message::message_header_view_mut::IggyMessageHeaderViewMut,
-};
-use lending_iterator::prelude::*;
-
-/// A mutable view of a message for in-place modifications
-#[derive(Debug)]
-pub struct IggyMessageViewMut<'a> {
- /// The buffer containing the message
- buffer: &'a mut [u8],
-}
-
-impl<'a> IggyMessageViewMut<'a> {
- /// Create a new mutable message view from a buffer
- pub fn new(buffer: &'a mut [u8]) -> Self {
- Self { buffer }
- }
-
- /// Get an immutable header view
- pub fn header(&self) -> IggyMessageHeaderView<'_> {
- let hdr_slice = &self.buffer[0..IGGY_MESSAGE_HEADER_SIZE];
- IggyMessageHeaderView::new(hdr_slice)
- }
-
- /// Get an ephemeral mutable header view for reading/writing
- pub fn header_mut(&mut self) -> IggyMessageHeaderViewMut<'_> {
- let hdr_slice = &mut self.buffer[0..IGGY_MESSAGE_HEADER_SIZE];
- IggyMessageHeaderViewMut::new(hdr_slice)
- }
-
- /// Returns the size of the entire message (header + payload + user headers).
- pub fn size(&self) -> usize {
- let hdr_view = self.header();
-
- IGGY_MESSAGE_HEADER_SIZE + hdr_view.payload_length() + hdr_view.user_headers_length()
- }
-
- /// Convenience method to update the checksum field in the header
- pub fn update_checksum(&mut self) {
- let checksum_field_size = size_of::<u64>(); // Skip checksum field for checksum calculation
- let size = self.size() - checksum_field_size;
- let data = &self.buffer[checksum_field_size..checksum_field_size + size];
- let checksum = calculate_checksum(data);
- self.header_mut().set_checksum(checksum);
- }
-}
-
-/// Iterator over mutable message views in a buffer
-pub struct IggyMessageViewMutIterator<'a> {
- buffer: &'a mut [u8],
- position: usize,
-}
-
-impl<'a> IggyMessageViewMutIterator<'a> {
- pub fn new(buffer: &'a mut [u8]) -> Self {
- Self {
- buffer,
- position: 0,
- }
- }
-}
-
-#[gat]
-impl LendingIterator for IggyMessageViewMutIterator<'_> {
- type Item<'next> = IggyMessageViewMut<'next>;
-
- fn next(&mut self) -> Option<Self::Item<'_>> {
- let buffer_len = self.buffer.len();
- if self.position >= buffer_len {
- return None;
- }
-
- if self.position + IGGY_MESSAGE_HEADER_SIZE > self.buffer.len() {
- tracing::error!(
- "Buffer too small for message header at position {}, buffer len: {}",
- self.position,
- self.buffer.len()
- );
- self.position = self.buffer.len();
- return None;
- }
-
- let buffer_slice = &mut self.buffer[self.position..];
- let view = IggyMessageViewMut::new(buffer_slice);
-
- let message_size = view.size();
- if message_size == 0 {
- tracing::error!(
- "Message size is 0 at position {}, preventing infinite loop",
- self.position
- );
- self.position = buffer_len;
- return None;
- }
-
- self.position += message_size;
- Some(view)
- }
-}
diff --git a/core/common/src/types/message/mod.rs b/core/common/src/types/message/mod.rs
index ccba3df11..86f236054 100644
--- a/core/common/src/types/message/mod.rs
+++ b/core/common/src/types/message/mod.rs
@@ -24,7 +24,6 @@ mod message_header;
mod message_header_view;
mod message_header_view_mut;
mod message_view;
-mod message_view_mut;
mod messages_batch;
pub mod partitioning;
pub mod partitioning_kind;
@@ -55,7 +54,6 @@ pub use message_header::{
pub use message_header_view::IggyMessageHeaderView;
pub use message_header_view_mut::IggyMessageHeaderViewMut;
pub use message_view::{IggyMessageView, IggyMessageViewIterator};
-pub use message_view_mut::{IggyMessageViewMut, IggyMessageViewMutIterator};
pub use messages_batch::IggyMessagesBatch;
pub use partitioning::Partitioning;
pub use partitioning_kind::PartitioningKind;
diff --git a/core/common/src/types/message/polled_messages.rs b/core/common/src/types/message/polled_messages.rs
index 6c45d54e2..017b8f539 100644
--- a/core/common/src/types/message/polled_messages.rs
+++ b/core/common/src/types/message/polled_messages.rs
@@ -15,8 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{IGGY_MESSAGE_HEADER_SIZE, IggyMessage, IggyMessageHeader, error::IggyError};
+use crate::{IggyMessage, IggyMessageHeader, error::IggyError};
use bytes::Bytes;
+use iggy_binary_protocol::batch::{BATCH_HEADER_SIZE, BatchHeader, BatchMessageHeader};
use serde::{Deserialize, Serialize};
use tracing::error;
@@ -50,6 +51,13 @@ impl PolledMessages {
}
impl PolledMessages {
+ /// Decode a `PollMessages` response body: the 16-byte prefix followed by
+ /// the served batch records (`[256B batch header][frames]`, deltas
+ /// resolved against the stamped bases).
+ ///
+ /// # Errors
+ /// [`IggyError::InvalidNumberEncoding`] on a short prefix;
+ /// [`IggyError::InvalidMessagePayloadLength`] on a malformed record.
pub fn from_bytes(bytes: Bytes) -> Result<Self, IggyError> {
let partition_id = u32::from_le_bytes(
bytes[0..4]
@@ -67,7 +75,7 @@ impl PolledMessages {
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
- let messages = messages_from_bytes_and_count(bytes.slice(16..), count)?;
+ let messages = messages_from_batches(bytes.slice(16..), count)?;
Ok(Self {
partition_id,
@@ -78,48 +86,61 @@ impl PolledMessages {
}
}
-/// Convert Bytes to messages
-fn messages_from_bytes_and_count(buffer: Bytes, count: u32) -> Result<Vec<IggyMessage>, IggyError> {
+/// Walk the served batch records, resolving each frame's deltas to absolute
+/// values. Payload and user-header `Bytes` are zero-copy slices of the
+/// response buffer.
+fn messages_from_batches(buffer: Bytes, count: u32) -> Result<Vec<IggyMessage>, IggyError> {
let mut messages = Vec::with_capacity(count as usize);
- let mut position = 0;
- let buf_len = buffer.len();
- while position < buf_len {
- if position + IGGY_MESSAGE_HEADER_SIZE > buf_len {
- break;
- }
- let header_bytes = buffer.slice(position..position + IGGY_MESSAGE_HEADER_SIZE);
- let header = match IggyMessageHeader::from_bytes(header_bytes) {
- Ok(h) => h,
- Err(e) => {
- error!("Failed to deserialize message header: {}", e);
- return Err(e);
+ let mut position = 0usize;
+ while position < buffer.len() {
+ let batch = BatchHeader::decode(&buffer[position..]).map_err(|decode_error| {
+ error!("Failed to decode polled batch header: {decode_error}");
+ IggyError::InvalidMessagePayloadLength
+ })?;
+ let batch_end = position
+ .checked_add(batch.total_size())
+ .filter(|end| *end <= buffer.len())
+ .ok_or(IggyError::InvalidMessagePayloadLength)?;
+ let mut cursor = position + BATCH_HEADER_SIZE;
+ while cursor < batch_end {
+ let frame =
+ BatchMessageHeader::decode(&buffer[cursor..batch_end]).map_err(|decode_error| {
+ error!("Failed to decode polled message frame: {decode_error}");
+ IggyError::InvalidMessagePayloadLength
+ })?;
+ let payload_start = cursor + iggy_binary_protocol::batch::BATCH_MESSAGE_HEADER_SIZE;
+ let payload_end = payload_start + frame.payload_length as usize;
+ let user_headers_end = payload_end + frame.user_headers_length as usize;
+ if user_headers_end > batch_end {
+ return Err(IggyError::InvalidMessagePayloadLength);
}
- };
- position += IGGY_MESSAGE_HEADER_SIZE;
-
- let payload_end = position + header.payload_length as usize;
- if payload_end > buf_len {
- break;
- }
- let payload = buffer.slice(position..payload_end);
- position = payload_end;
- let user_headers_end = position + header.user_headers_length as usize;
- if user_headers_end > buf_len {
- break;
+ let header = IggyMessageHeader {
+ checksum: frame.checksum,
+ id: frame.id,
+ offset: batch.base_offset + u64::from(frame.offset_delta),
+ // Broker append time is stamped once per batch; the
+ // per-message delta applies to `origin_timestamp` only.
+ timestamp: batch.base_timestamp,
+ origin_timestamp: batch.origin_timestamp + u64::from(frame.timestamp_delta),
+ user_headers_length: frame.user_headers_length,
+ payload_length: frame.payload_length,
+ reserved: 0,
+ };
+ let payload = buffer.slice(payload_start..payload_end);
+ let user_headers = if frame.user_headers_length > 0 {
+ Some(buffer.slice(payload_end..user_headers_end))
+ } else {
+ None
+ };
+ messages.push(IggyMessage {
+ header,
+ payload,
+ user_headers,
+ });
+ cursor = user_headers_end;
}
- let user_headers = if header.user_headers_length > 0 {
- Some(buffer.slice(position..user_headers_end))
- } else {
- None
- };
- position = user_headers_end;
-
- messages.push(IggyMessage {
- header,
- payload,
- user_headers,
- });
+ position = batch_end;
}
Ok(messages)
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 2db5aa0ed..3604e517b 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -70,7 +70,7 @@ use server_common::{
MESSAGE_ALIGN, Message, SegmentStorage,
iobuf::{Frozen, Owned},
send_messages::{
- ChecksumMode, SendMessagesHeader, convert_request_message, decode_prepare_slice,
+ BatchHeader, ChecksumMode, convert_request_message, decode_prepare_slice,
decode_prepare_slice_trusted, stamp_prepare_for_persistence,
},
sharding::IggyNamespace,
@@ -2652,7 +2652,7 @@ where
async fn append_stamped_messages(
&mut self,
message: Message<PrepareHeader>,
- batch: SendMessagesHeader,
+ batch: BatchHeader,
) -> Result<JournaledMessages, IggyError> {
let batch_messages_count = batch.message_count;
if batch_messages_count == 0 {
@@ -3639,7 +3639,6 @@ where
let segment_index = self.log.segments().len() - 1;
let segment = &mut self.log.segments_mut()[segment_index];
segment.size = IggyByteSize::from(segment.size.as_bytes_u64() + saved_bytes as u64);
- self.log.clear_in_flight();
return Ok(());
}
@@ -3691,7 +3690,6 @@ where
let segment = &mut self.log.segments_mut()[segment_index];
segment.size = IggyByteSize::from(segment.size.as_bytes_u64() + saved.as_bytes_u64());
- self.log.clear_in_flight();
Ok(())
}
@@ -3735,17 +3733,9 @@ where
},
);
- let storage = SegmentStorage::new(
- &messages_path,
- &index_path,
- 0,
- 0,
- enforce_fsync,
- enforce_fsync,
- false,
- )
- .await
- .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
+ let storage = SegmentStorage::new(&messages_path, &index_path, 0, 0, false)
+ .await
+ .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
let messages_size_bytes = storage
.messages_writer
.as_ref()
@@ -4011,17 +4001,9 @@ where
let enforce_fsync = self.effective_enforce_fsync(config);
let preallocate_segments = self.effective_preallocate_segments(config);
let segment = Segment::new(start_offset, segment_size);
- let storage = SegmentStorage::new(
- &messages_path,
- &index_path,
- 0,
- 0,
- enforce_fsync,
- enforce_fsync,
- false,
- )
- .await
- .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
+ let storage = SegmentStorage::new(&messages_path, &index_path, 0, 0, false)
+ .await
+ .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
let messages_size_bytes = storage
.messages_writer
.as_ref()
@@ -6447,10 +6429,9 @@ mod tests {
let log_path = format!("{partition_dir}/{:0>20}.log", 0u64);
let index_path = format!("{partition_dir}/{:0>20}.index", 0u64);
partition.log.segments_mut()[0].sealed = true;
- partition.log.storages_mut()[0] =
- SegmentStorage::new(&log_path, &index_path, 0, 0, false, false, false)
- .await
- .expect("create segment storage");
+ partition.log.storages_mut()[0] = SegmentStorage::new(&log_path, &index_path, 0, 0, false)
+ .await
+ .expect("create segment storage");
let record = build_segment_record(namespace, 0);
let record_len = record.len() as u64;
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index fcb33324f..905ae6119 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -19,7 +19,7 @@ use iggy_binary_protocol::{Operation, PrepareHeader};
use journal::{Journal, Storage};
use server_common::{
iobuf::{Frozen, Owned},
- send_messages::{COMMAND_HEADER_SIZE, SendMessagesRef, decode_prepare_slice_trusted},
+ send_messages::{self, BatchRef, COMMAND_HEADER_SIZE, decode_prepare_slice_trusted},
};
use std::io;
use std::{
@@ -1023,7 +1023,7 @@ impl Journal<PartitionJournalMemStorage> for PartitionJournal<PartitionJournalMe
}
pub fn select_batch_slice(
- batch: &SendMessagesRef<'_>,
+ batch: &BatchRef<'_>,
query: MessageLookup,
already_matched: u32,
) -> Option<SelectedBatchSlice> {
@@ -1087,7 +1087,7 @@ pub fn select_batch_slice(
/// Push the fragments for one selected batch, shared by the resident-journal
/// walk and the disk-chunk walk. `source` holds a stamped
-/// `[256B SendMessagesHeader][blob]` batch starting at byte `batch_base`
+/// `[256B BatchHeader][blob]` batch starting at byte `batch_base`
/// (the disk walk passes the chunk cursor; the resident walk passes
/// `size_of::<PrepareHeader>()`, the batch's offset past the prepare header).
/// A full-body selection forwards the original batch bytes by reference; a
@@ -1099,7 +1099,7 @@ pub fn push_selected_batch_fragments(
matched_messages: &mut u32,
source: &Frozen<4096>,
batch_base: usize,
- batch: &SendMessagesRef<'_>,
+ batch: &BatchRef<'_>,
selection: SelectedBatchSlice,
) {
let full_body_selected = selection.start == 0 && selection.end == batch.blob().len();
@@ -1122,7 +1122,9 @@ pub fn push_selected_batch_fragments(
.get(selection.start..selection.end)
.expect("selected batch slice must stay within blob bounds"),
);
- fragments.push(Fragment::whole(rewritten.into_frozen()));
+ fragments.push(Fragment::whole(send_messages::frozen_batch_header(
+ &rewritten,
+ )));
fragments.push(Fragment::slice(
source.clone(),
batch_base + COMMAND_HEADER_SIZE + selection.start,
diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs
index 574b37108..fc3a7744b 100644
--- a/core/partitions/src/log.rs
+++ b/core/partitions/src/log.rs
@@ -20,10 +20,10 @@ use crate::iggy_index_writer::IggyIndexWriter;
use crate::messages_writer::MessagesWriter;
use crate::poll_plan::SealedSegmentHandle;
use crate::segment::Segment;
-use iggy_common::{IggyByteSize, IggyMessagesBatch};
+use iggy_common::IggyByteSize;
use journal::{Journal, Storage};
use ringbuffer::AllocRingBuffer;
-use server_common::{IggyMessagesBatchSetInFlight, SegmentStorage};
+use server_common::SegmentStorage;
use std::collections::VecDeque;
use std::fmt::Debug;
use std::rc::Rc;
@@ -166,7 +166,6 @@ where
// `SEALED_READ_STATE_CAP`. Keyed by offset (stable), not slot index (which
// shifts on retire). See `touch_sealed_read_state`.
sealed_lru: VecDeque<u64>,
- in_flight: IggyMessagesBatchSetInFlight,
}
impl<J, S> Default for SegmentedLog<J, S>
@@ -187,7 +186,6 @@ where
index_writers: Vec::with_capacity(SEGMENTS_CAPACITY),
sealed_read_state: Vec::with_capacity(SEGMENTS_CAPACITY),
sealed_lru: VecDeque::with_capacity(SEALED_READ_STATE_CAP + 1),
- in_flight: IggyMessagesBatchSetInFlight::default(),
}
}
}
@@ -431,22 +429,6 @@ where
*segment_indexes = Some(indexes);
}
}
-
- pub const fn in_flight(&self) -> &IggyMessagesBatchSetInFlight {
- &self.in_flight
- }
-
- pub const fn in_flight_mut(&mut self) -> &mut IggyMessagesBatchSetInFlight {
- &mut self.in_flight
- }
-
- pub fn set_in_flight(&mut self, batches: Vec<IggyMessagesBatch>) {
- self.in_flight.set(batches);
- }
-
- pub fn clear_in_flight(&mut self) {
- self.in_flight.clear();
- }
}
impl<J, S> SegmentedLog<J, S>
diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs
index 2753b42b5..3b19f0dbb 100644
--- a/core/partitions/src/poll_plan.rs
+++ b/core/partitions/src/poll_plan.rs
@@ -455,7 +455,7 @@ pub enum DiskReadOutcome {
impl DiskReadPlan {
/// Serve a poll from the on-disk segment files, off the partition borrow.
/// Reads from owned descriptors so no partition reference is held across
- /// the file IO. Walks stamped `[256B SendMessagesHeader][blob]` batches in
+ /// the file IO. Walks stamped `[256B BatchHeader][blob]` batches in
/// chunked reads, re-reading a batch split across a chunk boundary in the
/// next chunk.
#[allow(clippy::cast_possible_truncation)]
@@ -901,7 +901,7 @@ pub fn upsert_offset_max<K>(
}
}
-/// Walk stamped `[256B SendMessagesHeader][blob]` batches in one disk
+/// Walk stamped `[256B BatchHeader][blob]` batches in one disk
/// chunk, pushing matching fragments. Returns bytes consumed: the start
/// of the first batch that did not fully fit in the chunk (the caller
/// re-reads from there), or the chunk end when everything decoded.
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 0102985dc..1653b1913 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -2263,18 +2263,8 @@ where
// sweep itself is right (a chain the live state does not know
// about would resurrect at boot), so one retry against a
// transient open failure is the only cheap save available.
- let enforce_fsync = self.effective_enforce_fsync(config);
- let open = || {
- SegmentStorage::new(
- &log_final,
- &index_final,
- meta.size,
- meta.index_size,
- enforce_fsync,
- enforce_fsync,
- true,
- )
- };
+ let open =
+ || SegmentStorage::new(&log_final, &index_final, meta.size, meta.index_size, true);
let storage = match open().await {
Ok(storage) => storage,
Err(_) => open()
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 8b1613577..7895d6ad2 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -2437,16 +2437,12 @@ async fn recover_partition_segments(
let segment_size = runtime_options
.segment_size
.unwrap_or_else(|| IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE));
- let enforce_fsync = runtime_options
- .enforce_fsync
- .unwrap_or(iggy_common::DEFAULT_ENFORCE_FSYNC);
load_persisted_segments(
config,
stream_id,
topic_id,
partition_id,
segment_size,
- enforce_fsync,
stats,
)
.await
diff --git a/core/server/src/http/wire.rs b/core/server/src/http/wire.rs
index b6cb697b0..5048bc428 100644
--- a/core/server/src/http/wire.rs
+++ b/core/server/src/http/wire.rs
@@ -77,7 +77,13 @@ pub(in crate::http) fn encode_send_messages(
&wire_topic_id,
&wire_partitioning,
&raw_messages,
- );
+ )
+ .map_err(|error| match error {
+ iggy_binary_protocol::WireError::InvalidMessageTimestampDelta(delta) => {
+ IggyError::InvalidMessageTimestampDelta(delta)
+ }
+ _ => IggyError::InvalidCommand,
+ })?;
Ok(buf.freeze())
}
@@ -214,8 +220,6 @@ mod tests {
use iggy_binary_protocol::WireDecode;
use iggy_binary_protocol::WireEncode;
- use iggy_binary_protocol::WireMessageIterator;
- use iggy_binary_protocol::message_layout::WIRE_MESSAGE_INDEX_SIZE;
use iggy_common::delete_consumer_offset::DeleteConsumerOffset;
use iggy_common::{
Consumer, ConsumerKind, IggyMessagesBatch, IggyTimestamp, Partitioning, PartitioningKind,
@@ -225,7 +229,7 @@ mod tests {
use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Owned;
use server_common::send_messages::{
- COMMAND_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesHeader,
+ BatchHeader, COMMAND_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages,
SendMessagesOwned,
};
@@ -286,18 +290,26 @@ mod tests {
);
assert_eq!(header.messages_count, 2);
- let data_offset = 4 + metadata_length + 2 * WIRE_MESSAGE_INDEX_SIZE;
- let views: Vec<_> = WireMessageIterator::new(&bytes[data_offset..], 2)
- .collect::<Result<Vec<_>, _>>()
- .expect("valid message frames");
- assert_eq!(views[0].id(), 7);
- assert_eq!(views[0].payload(), b"first");
- assert_eq!(views[0].user_headers(), b"");
- assert_eq!(views[0].origin_timestamp(), origin_timestamps[0]);
- assert_eq!(views[1].id(), 8);
- assert_eq!(views[1].payload(), b"second");
- assert_eq!(views[1].user_headers(), b"raw-header-bytes");
- assert_eq!(views[1].origin_timestamp(), origin_timestamps[1]);
+ let batch = iggy_binary_protocol::batch::decode_batch_slice(&bytes[4 + metadata_length..])
+ .expect("valid producer batch");
+ assert_eq!(batch.header.partition_id, 0);
+ assert_eq!(batch.message_count(), 2);
+ let views: Vec<_> = batch.iter().collect();
+ let batch_origin = batch.header.origin_timestamp;
+ assert_eq!(views[0].header.id, 7);
+ assert_eq!(views[0].payload, b"first");
+ assert_eq!(views[0].user_headers, b"");
+ assert_eq!(
+ batch_origin + u64::from(views[0].header.timestamp_delta),
+ origin_timestamps[0]
+ );
+ assert_eq!(views[1].header.id, 8);
+ assert_eq!(views[1].payload, b"second");
+ assert_eq!(views[1].user_headers, b"raw-header-bytes");
+ assert_eq!(
+ batch_origin + u64::from(views[1].header.timestamp_delta),
+ origin_timestamps[1]
+ );
}
#[test]
@@ -469,7 +481,7 @@ mod tests {
/// Wrap one stored `SendMessages` batch (`[256B header][blob]`) as the
/// poll fragment the owning shard replies, the shape
/// `build_polled_messages_body` consumes.
- fn fragment_from_stored_batch(header: &SendMessagesHeader, blob: &[u8]) -> PollFragments {
+ fn fragment_from_stored_batch(header: &BatchHeader, blob: &[u8]) -> PollFragments {
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(COMMAND_HEADER_SIZE + blob.len());
header.encode_into(buffer.as_mut_slice());
buffer.as_mut_slice()[COMMAND_HEADER_SIZE..].copy_from_slice(blob);
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index e3117fdd3..c9d968fed 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -377,26 +377,18 @@ pub async fn ensure_initial_segment(
// this with a stale `.index` at offset 0 on disk. The `partitions`-side
// writers with the same names do NOT truncate, so opening them directly
// instead would read index entries from a previous generation.
- let storage = SegmentStorage::new(
- &messages_path,
- &index_path,
- 0,
- 0,
- enforce_fsync,
- enforce_fsync,
- false,
- )
- .await
- .map_err(|source| {
- error!(
- stream_id,
- topic_id,
- partition_id,
- error = %source,
- "failed to create initial segment storage"
- );
- source
- })?;
+ let storage = SegmentStorage::new(&messages_path, &index_path, 0, 0, false)
+ .await
+ .map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ error = %source,
+ "failed to create initial segment storage"
+ );
+ source
+ })?;
// Share the storage's size counters so reads observe persisted bytes;
// a writer with a private counter grows the file invisibly to readers.
let messages_size_counter = storage
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 02e9a3647..6b7fc4ec5 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -91,7 +91,7 @@ use journal::{Journal, JournalHandle};
use metadata::impls::metadata::StreamsFrontend;
use partitions::PollFragments;
use server_common::Message;
-use server_common::send_messages::{self, COMMAND_HEADER_SIZE};
+use server_common::send_messages;
use server_common::sharding::IggyNamespace;
use shard::ConnectedClientInfo;
use std::cell::RefCell;
@@ -1615,24 +1615,17 @@ where
.map_or(0, VsrConsensus::commit_max)
}
-/// Size of the in-storage (`IggyMessage`) per-message header inside a
-/// `SendMessages` batch blob: `checksum`(8) + `id`(16) + `offset_delta`(4)
-/// + `timestamp_delta`(4) + `user_headers_length`(4) + `payload_length`(4)
-/// + reserved(8). See `server_common::send_messages::SendMessagesOwned::from_messages`.
-const STORED_MESSAGE_HEADER_SIZE: usize = 48;
-
/// Build the `PolledMessages` reply body from the owning shard's poll
/// fragments.
///
-/// Fragments carry the stored `SendMessages` batches: a 256-byte command
-/// header followed by `IggyMessage`-format messages
-/// (`[48B header][payload][user_headers]`, offsets/timestamps delta-encoded
-/// against the batch). The SDK decodes the legacy wire format
-/// (`[64B header][payload][user_headers]`, absolute offsets); the message
-/// sections share the legacy order, so only the header is re-encoded here
-/// and the section bytes copy through contiguously.
+/// Fragments carry the stored batch records (a 256-byte batch header plus
+/// `[48B header][payload][user_headers]` frames, deltas resolved against the
+/// stamped bases) and are served to the client as they are - the reply's
+/// message encoding IS the storage encoding. The one rewrite left is at-rest
+/// decryption: stored sections are ciphertext, and this reply is the single
+/// decrypt point, so encrypted records are rebuilt over the plaintext.
///
-/// Body layout: `[partition_id:4][current_offset:8][count:4][messages...]`.
+/// Body layout: `[partition_id:4][current_offset:8][count:4][batch records...]`.
pub(crate) fn build_polled_messages_body(
partition_id: u32,
current_offset: u64,
@@ -1643,17 +1636,14 @@ pub(crate) fn build_polled_messages_body(
// COUNT_OFFSET and is backpatched once the walk below knows it.
const HEAD_LEN: usize = 16;
const COUNT_OFFSET: usize = 12;
- // Batches may arrive split across fragments (rewritten command header +
- // sliced blob); concatenate into one stream before walking batches.
+ // Batches may arrive split across fragments (rewritten batch header +
+ // sliced blob); concatenate into one stream before walking records.
let mut stream: Vec<u8> = Vec::new();
for fragment in fragments {
let frozen = fragment.into_frozen();
stream.extend_from_slice(frozen.as_slice());
}
- // Reserve the head and encode the messages straight into `body`; encoding
- // directly avoids a separate messages buffer and its copy-through into the
- // head.
let mut body: Vec<u8> = Vec::with_capacity(HEAD_LEN + stream.len());
body.extend_from_slice(&partition_id.to_le_bytes());
body.extend_from_slice(¤t_offset.to_le_bytes());
@@ -1661,104 +1651,24 @@ pub(crate) fn build_polled_messages_body(
let mut count: u32 = 0;
let mut position = 0usize;
while position < stream.len() {
- let batch = send_messages::SendMessagesHeader::decode(&stream[position..])?;
+ let batch = send_messages::BatchHeader::decode(&stream[position..])
+ .map_err(|_| IggyError::InvalidCommand)?;
let batch_end = position
- .checked_add(
- usize::try_from(batch.batch_length).map_err(|_| IggyError::InvalidCommand)?,
- )
+ .checked_add(batch.total_size())
.ok_or(IggyError::InvalidCommand)?;
if batch_end > stream.len() {
return Err(IggyError::InvalidCommand);
}
- let mut cursor = position + COMMAND_HEADER_SIZE;
- while cursor < batch_end {
- if cursor + STORED_MESSAGE_HEADER_SIZE > batch_end {
- return Err(IggyError::InvalidCommand);
- }
- let header = &stream[cursor..cursor + STORED_MESSAGE_HEADER_SIZE];
- let checksum = &header[0..8];
- let id = &header[8..24];
- let offset_delta = u32::from_le_bytes(header[24..28].try_into().expect("4-byte slice"));
- let timestamp_delta =
- u32::from_le_bytes(header[28..32].try_into().expect("4-byte slice"));
- let user_headers_length =
- u32::from_le_bytes(header[32..36].try_into().expect("4-byte slice")) as usize;
- let payload_length =
- u32::from_le_bytes(header[36..40].try_into().expect("4-byte slice")) as usize;
-
- let sections_start = cursor + STORED_MESSAGE_HEADER_SIZE;
- let sections_end = sections_start + payload_length + user_headers_length;
- if sections_end > batch_end {
- return Err(IggyError::InvalidCommand);
- }
-
- let offset = batch.base_offset + u64::from(offset_delta);
- // `base_timestamp` is the flat broker append time stamped once per
- // batch; `timestamp_delta` is a per-message delta against the
- // producer origin, so it only applies to `origin_timestamp`. Adding
- // it to the broker base would mix two clocks.
- let timestamp = batch.base_timestamp;
- let origin_timestamp = batch.origin_timestamp + u64::from(timestamp_delta);
-
- body.extend_from_slice(checksum);
- body.extend_from_slice(id);
- body.extend_from_slice(&offset.to_le_bytes());
- body.extend_from_slice(×tamp.to_le_bytes());
- body.extend_from_slice(&origin_timestamp.to_le_bytes());
- if let Some(encryptor) = encryptor {
- // At-rest encryption: stored sections are ciphertext (encrypted
- // once at ingestion, replicated verbatim); this reply is the
- // single decrypt point, so lengths are rewritten to the
- // plaintext sizes. The stored per-message checksum still covers
- // the ciphertext and is passed through untouched (the SDK does
- // not re-validate it against the reply layout).
- let payload_end = sections_start + payload_length;
- let payload = encryptor
- .decrypt(&stream[sections_start..payload_end])
- .map_err(|_| IggyError::CannotDecryptData)?;
- let user_headers = if user_headers_length > 0 {
- Some(
- encryptor
- .decrypt(&stream[payload_end..sections_end])
- .map_err(|_| IggyError::CannotDecryptData)?,
- )
- } else {
- None
- };
- let user_headers_bytes: &[u8] = user_headers.as_deref().unwrap_or_default();
- body.extend_from_slice(
- &u32::try_from(user_headers_bytes.len())
- .map_err(|_| IggyError::InvalidCommand)?
- .to_le_bytes(),
- );
- body.extend_from_slice(
- &u32::try_from(payload.len())
- .map_err(|_| IggyError::InvalidCommand)?
- .to_le_bytes(),
- );
- body.extend_from_slice(&0u64.to_le_bytes()); // reserved
- body.extend_from_slice(&payload);
- body.extend_from_slice(user_headers_bytes);
- } else {
- body.extend_from_slice(
- &u32::try_from(user_headers_length)
- .expect("length came from u32")
- .to_le_bytes(),
- );
- body.extend_from_slice(
- &u32::try_from(payload_length)
- .expect("length came from u32")
- .to_le_bytes(),
- );
- body.extend_from_slice(&0u64.to_le_bytes()); // reserved
- // Stored sections are already in legacy order
- // (`[payload][user_headers]`): copy through contiguously.
- body.extend_from_slice(&stream[sections_start..sections_end]);
- }
-
- count += 1;
- cursor = sections_end;
+ let record = &stream[position..batch_end];
+ if let Some(encryptor) = encryptor {
+ let decrypted = send_messages::decrypt_batch_record(record, encryptor)?;
+ body.extend_from_slice(&decrypted);
+ } else {
+ body.extend_from_slice(record);
}
+ count = count
+ .checked_add(batch.message_count)
+ .ok_or(IggyError::InvalidCommand)?;
position = batch_end;
}
diff --git a/core/server/src/segment_recovery.rs b/core/server/src/segment_recovery.rs
index 9c639f8f7..fcaa57ace 100644
--- a/core/server/src/segment_recovery.rs
+++ b/core/server/src/segment_recovery.rs
@@ -33,7 +33,7 @@ use iggy_common::{IggyByteSize, IggyError, PartitionStats};
use partitions::state_transfer::STAGING_SUFFIX;
use partitions::{IggyIndexReader, Segment};
use server_common::SegmentStorage;
-use server_common::send_messages::{COMMAND_HEADER_SIZE, SendMessagesHeader, decode_batch_slice};
+use server_common::send_messages::{BatchHeader, COMMAND_HEADER_SIZE, decode_batch_slice};
use std::fs;
use std::os::unix::fs::FileExt;
use std::path::PathBuf;
@@ -66,7 +66,6 @@ pub async fn load_persisted_segments(
topic_id: usize,
partition_id: usize,
segment_size: IggyByteSize,
- enforce_fsync: bool,
stats: &PartitionStats,
) -> Result<Vec<RecoveredSegment>, ServerError> {
let partition_path = config
@@ -140,8 +139,6 @@ pub async fn load_persisted_segments(
&index_path,
effective_messages_size,
effective_index_size,
- enforce_fsync,
- enforce_fsync,
true,
)
.await
@@ -521,7 +518,7 @@ async fn recover_segment_bounds(
fn batch_verifies(
messages: &fs::File,
position: u64,
- header: &SendMessagesHeader,
+ header: &BatchHeader,
scratch: &mut Vec<u8>,
) -> bool {
scratch.clear();
@@ -536,11 +533,11 @@ fn read_batch_header(
messages: &fs::File,
position: u64,
messages_size: u64,
-) -> Option<SendMessagesHeader> {
+) -> Option<BatchHeader> {
if position.checked_add(COMMAND_HEADER_SIZE as u64)? > messages_size {
return None;
}
let mut header_bytes = [0u8; COMMAND_HEADER_SIZE];
messages.read_exact_at(&mut header_bytes, position).ok()?;
- SendMessagesHeader::decode(&header_bytes).ok()
+ BatchHeader::decode(&header_bytes).ok()
}
diff --git a/core/server_common/src/in_flight.rs b/core/server_common/src/in_flight.rs
deleted file mode 100644
index 946642958..000000000
--- a/core/server_common/src/in_flight.rs
+++ /dev/null
@@ -1,100 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use iggy_common::IggyMessagesBatch;
-
-/// Holds batches that are being written to disk.
-///
-/// During async I/O, messages are transferred from the journal to disk.
-/// This buffer holds frozen (immutable, Arc-backed) copies so consumers
-/// can still read them during the write operation.
-#[derive(Debug, Default)]
-pub struct IggyMessagesBatchSetInFlight {
- batches: Vec<IggyMessagesBatch>,
- first_offset: u64,
- last_offset: u64,
-}
-
-impl IggyMessagesBatchSetInFlight {
- pub fn is_empty(&self) -> bool {
- self.batches.is_empty()
- }
-
- pub fn first_offset(&self) -> u64 {
- self.first_offset
- }
-
- pub fn last_offset(&self) -> u64 {
- self.last_offset
- }
-
- pub fn set(&mut self, batches: Vec<IggyMessagesBatch>) {
- if batches.is_empty() {
- self.clear();
- return;
- }
- self.first_offset = batches.first().and_then(|b| b.first_offset()).unwrap_or(0);
- self.last_offset = batches.last().and_then(|b| b.last_offset()).unwrap_or(0);
- self.batches = batches;
- }
-
- pub fn clear(&mut self) {
- self.batches.clear();
- self.first_offset = 0;
- self.last_offset = 0;
- }
-
- /// Get batches that overlap with the requested offset range.
- /// Returns only the batches that contain messages within [start_offset, start_offset + count).
- pub fn get_by_offset(&self, start_offset: u64, count: u32) -> &[IggyMessagesBatch] {
- if self.is_empty() || start_offset > self.last_offset || count == 0 {
- return &[];
- }
-
- let end_offset = start_offset + count as u64 - 1;
- if end_offset < self.first_offset {
- return &[];
- }
-
- let start_idx = self
- .batches
- .iter()
- .position(|b| {
- b.last_offset()
- .map(|last| last >= start_offset)
- .unwrap_or(false)
- })
- .unwrap_or(0);
-
- let end_idx = self
- .batches
- .iter()
- .rposition(|b| {
- b.first_offset()
- .map(|first| first <= end_offset)
- .unwrap_or(false)
- })
- .map(|i| i + 1)
- .unwrap_or(self.batches.len());
-
- &self.batches[start_idx..end_idx]
- }
-
- pub fn batches(&self) -> &[IggyMessagesBatch] {
- &self.batches
- }
-}
diff --git a/core/server_common/src/indexes_mut.rs b/core/server_common/src/indexes_mut.rs
deleted file mode 100644
index 394b234cb..000000000
--- a/core/server_common/src/indexes_mut.rs
+++ /dev/null
@@ -1,388 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use crate::PooledBuffer;
-use iggy_common::{INDEX_SIZE, IggyIndexView, IggyIndexes};
-use std::fmt;
-use std::ops::{Deref, Index as StdIndex};
-
-/// A container for binary-encoded index data.
-/// Optimized for efficient storage and I/O operations.
-#[derive(Default)]
-pub struct IggyIndexesMut {
- buffer: PooledBuffer,
- saved_count: u32,
- base_position: u32,
-}
-
-impl IggyIndexesMut {
- /// Creates a new empty container
- pub fn empty() -> Self {
- Self {
- buffer: PooledBuffer::empty(),
- saved_count: 0,
- base_position: 0,
- }
- }
-
- /// Creates indexes from bytes
- pub fn from_bytes(indexes: PooledBuffer, base_position: u32) -> Self {
- Self {
- buffer: indexes,
- saved_count: 0,
- base_position,
- }
- }
-
- /// Decompose the container into its components
- pub fn decompose(mut self) -> (u32, PooledBuffer) {
- let base_position = self.base_position;
- let buffer = std::mem::take(&mut self.buffer);
- (base_position, buffer)
- }
-
- /// Freezes the indexes buffer, converting to an immutable `IggyIndexes`.
- ///
- /// The returned `IggyIndexes` uses Arc-backed `Bytes`, allowing cheap clones.
- pub fn freeze(&mut self) -> IggyIndexes {
- let base_position = self.base_position;
- let buffer = self.buffer.freeze();
- self.saved_count = 0;
- self.base_position = 0;
- IggyIndexes::new(buffer, base_position)
- }
-
- /// Gets the size of all indexes messages
- pub fn messages_size(&self) -> u32 {
- self.last_position() - self.base_position
- }
-
- /// Gets the base position of the indexes
- pub fn base_position(&self) -> u32 {
- self.base_position
- }
-
- /// Sets the base position of the indexes
- pub fn set_base_position(&mut self, base_position: u32) {
- self.base_position = base_position;
- }
-
- /// Helper method to get the last index position
- pub fn last_position(&self) -> u32 {
- self.get(self.count() - 1)
- .map(|idx| idx.position())
- .unwrap_or(0)
- }
-
- /// Creates a new container with the specified capacity
- pub fn with_capacity(capacity: usize, base_position: u32) -> Self {
- Self {
- buffer: PooledBuffer::with_capacity(capacity * INDEX_SIZE),
- saved_count: 0,
- base_position,
- }
- }
-
- /// Gets the capacity of the buffer
- pub fn capacity(&self) -> usize {
- self.buffer.capacity()
- }
-
- /// Inserts a new index at the end of buffer
- pub fn insert(&mut self, offset: u32, position: u32, timestamp: u64) {
- self.buffer.put_u32_le(offset);
- self.buffer.put_u32_le(position);
- self.buffer.put_u64_le(timestamp);
- }
-
- /// Appends another slice of indexes to this one.
- pub fn append_slice(&mut self, other: &[u8]) {
- self.buffer.put_slice(other);
- }
-
- /// Gets the number of indexes in the container
- pub fn count(&self) -> u32 {
- self.buffer.len() as u32 / INDEX_SIZE as u32
- }
-
- /// Checks if the container is empty
- pub fn is_empty(&self) -> bool {
- self.count() == 0
- }
-
- /// Gets the size of the buffer in bytes
- pub fn size(&self) -> u32 {
- self.buffer.len() as u32
- }
-
- /// Gets a view of the Index at the specified index
- pub fn get(&self, index: u32) -> Option<IggyIndexView<'_>> {
- if index >= self.count() {
- return None;
- }
-
- let start = index as usize * INDEX_SIZE;
- let end = start + INDEX_SIZE;
-
- if end <= self.buffer.len() {
- Some(IggyIndexView::new(&self.buffer[start..end]))
- } else {
- None
- }
- }
-
- // Set the offset at the given index position
- pub fn set_offset_at(&mut self, index: u32, offset: u32) {
- let pos = index as usize * INDEX_SIZE;
- self.buffer[pos..pos + 4].copy_from_slice(&offset.to_le_bytes());
- }
-
- // Set the position at the given index
- pub fn set_position_at(&mut self, index: u32, position: u32) {
- let pos = (index as usize * INDEX_SIZE) + 4;
- self.buffer[pos..pos + 4].copy_from_slice(&position.to_le_bytes());
- }
-
- // Set the timestamp at the given index
- pub fn set_timestamp_at(&mut self, index: u32, timestamp: u64) {
- let pos = (index as usize * INDEX_SIZE) + 8;
- self.buffer[pos..pos + 8].copy_from_slice(×tamp.to_le_bytes());
- }
-
- /// Gets a last index
- pub fn last(&self) -> Option<IggyIndexView<'_>> {
- if self.count() == 0 {
- return None;
- }
-
- // Bound to exactly one entry rather than to the buffer end: a file
- // whose length is not a whole multiple of INDEX_SIZE (e.g. a 24-byte
- // server sparse index read back through this 16-byte reader on a
- // mixed-format recovery) would otherwise hand an oversized slice to the
- // view and trip its length assertion.
- let start = (self.count() - 1) as usize * INDEX_SIZE;
- Some(IggyIndexView::new(&self.buffer[start..start + INDEX_SIZE]))
- }
-
- /// Finds an index by timestamp using binary search
- /// If an exact match isn't found, returns the index with the nearest timestamp
- /// that is greater than or equal to the requested timestamp
- pub fn find_by_timestamp(&self, timestamp: u64) -> Option<IggyIndexView<'_>> {
- if self.count() == 0 {
- return None;
- }
-
- let first_idx = self.get(0)?;
- if timestamp <= first_idx.timestamp() {
- return Some(first_idx);
- }
-
- let last_saved_idx = self.get(self.count() - 1)?;
- if timestamp > last_saved_idx.timestamp() {
- return None;
- }
-
- let mut left = 0;
- let mut right = self.count() as isize - 1;
- let mut result: Option<IggyIndexView<'_>> = None;
-
- while left <= right {
- let mid = left + (right - left) / 2;
- let view = self.get(mid as u32).unwrap();
- let current_timestamp = view.timestamp();
-
- match current_timestamp.cmp(×tamp) {
- std::cmp::Ordering::Equal => {
- result = Some(view);
- right = mid - 1;
- }
- std::cmp::Ordering::Less => {
- left = mid + 1;
- }
- std::cmp::Ordering::Greater => {
- result = Some(view);
- right = mid - 1;
- }
- }
- }
-
- result
- }
-
- /// Clears the container, removing all indexes but preserving already allocated buffer capacity
- pub fn clear(&mut self) {
- self.saved_count = 0;
- self.buffer.clear();
- }
-
- /// Gets the unsaved part of the index buffer
- pub fn unsaved_slice(&self) -> PooledBuffer {
- let start_pos = self.saved_count as usize * INDEX_SIZE;
- // TODO: Dunno how to handle this better, maybe we should have a `split` method,
- // That splits the underlying Indexes buffer into two parts
- // saved on disk and not saved yet.
- PooledBuffer::from(&self.buffer[start_pos..])
- }
-
- /// Mark all indexes as saved to disk
- pub fn mark_saved(&mut self) {
- self.saved_count = self.count();
- }
-
- /// Slices the container to return a view of a specific range of indexes
- pub fn slice_by_offset(
- &self,
- relative_start_offset: u32,
- count: u32,
- ) -> Option<IggyIndexesMut> {
- let available_count = self.count().saturating_sub(relative_start_offset);
- let actual_count = std::cmp::min(count, available_count);
-
- if actual_count == 0 || relative_start_offset >= self.count() {
- return None;
- }
-
- let end_pos = relative_start_offset + actual_count;
-
- let start_byte = relative_start_offset as usize * INDEX_SIZE;
- let end_byte = end_pos as usize * INDEX_SIZE;
- let slice = PooledBuffer::from(&self.buffer[start_byte..end_byte]);
-
- if relative_start_offset == 0 {
- Some(IggyIndexesMut::from_bytes(slice, self.base_position))
- } else {
- let position_offset: u32 = self.get(relative_start_offset - 1).unwrap().position();
- Some(IggyIndexesMut::from_bytes(slice, position_offset))
- }
- }
-
- /// Loads indexes from cache based on timestamp
- pub fn slice_by_timestamp(&self, timestamp: u64, count: u32) -> Option<IggyIndexesMut> {
- if self.count() == 0 {
- return None;
- }
-
- let start_index_pos = self.binary_search_position_for_timestamp_sync(timestamp)?;
-
- let available_count = self.count().saturating_sub(start_index_pos);
- let actual_count = std::cmp::min(count, available_count);
-
- if actual_count == 0 {
- return None;
- }
-
- let end_pos = start_index_pos + actual_count;
-
- let start_byte = start_index_pos as usize * INDEX_SIZE;
- let end_byte = end_pos as usize * INDEX_SIZE;
- let slice = PooledBuffer::from(&self.buffer[start_byte..end_byte]);
-
- let base_position = if start_index_pos > 0 {
- self.get(start_index_pos - 1).unwrap().position()
- } else {
- 0
- };
-
- Some(IggyIndexesMut::from_bytes(slice, base_position))
- }
-
- /// Find the position of the index with timestamp closest to (but not exceeding) the target
- fn binary_search_position_for_timestamp_sync(&self, target_timestamp: u64) -> Option<u32> {
- if self.count() == 0 {
- return None;
- }
-
- let last_index = self.get(self.count() - 1)?;
- if target_timestamp > last_index.timestamp() {
- return Some(self.count() - 1);
- }
-
- let first_index = self.get(0)?;
- if target_timestamp <= first_index.timestamp() {
- return Some(0);
- }
-
- let mut low = 0;
- let mut high = self.count() - 1;
-
- while low <= high {
- let mid = low + (high - low) / 2;
- let mid_index = self.get(mid)?;
- let mid_timestamp = mid_index.timestamp();
-
- match mid_timestamp.cmp(&target_timestamp) {
- std::cmp::Ordering::Equal => return Some(mid),
- std::cmp::Ordering::Less => low = mid + 1,
- std::cmp::Ordering::Greater => {
- if mid == 0 {
- break;
- }
- high = mid - 1;
- }
- }
- }
-
- Some(low)
- }
-}
-
-impl StdIndex<usize> for IggyIndexesMut {
- type Output = [u8];
-
- fn index(&self, index: usize) -> &Self::Output {
- let start = index * INDEX_SIZE;
- let end = start + INDEX_SIZE;
- &self.buffer[start..end]
- }
-}
-
-impl Deref for IggyIndexesMut {
- type Target = [u8];
-
- fn deref(&self) -> &Self::Target {
- &self.buffer
- }
-}
-
-impl fmt::Debug for IggyIndexesMut {
- fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- let count = self.count();
-
- if count == 0 {
- return write!(f, "IggyIndexesMut {{ count: 0, indexes: [] }}");
- }
-
- writeln!(f, "IggyIndexesMut {{")?;
- writeln!(f, " count: {count},")?;
- writeln!(f, " indexes: [")?;
-
- for i in 0..count {
- if let Some(index) = self.get(i) {
- writeln!(
- f,
- " {{ offset: {}, position: {}, timestamp: {} }},",
- index.offset(),
- index.position(),
- index.timestamp()
- )?;
- }
- }
-
- writeln!(f, " ]")?;
- write!(f, "}}")
- }
-}
diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs
index bf4f9f9a2..2aca6f258 100644
--- a/core/server_common/src/lib.rs
+++ b/core/server_common/src/lib.rs
@@ -23,8 +23,6 @@ pub mod crypto;
pub mod diagnostics;
pub mod executor;
pub mod fs_utils;
-mod in_flight;
-mod indexes_mut;
// TODO(hubcio): iobuf was relocated verbatim from `core/binary_protocol/src/consensus/iobuf.rs`
// during the sans-io split. Its implementation is intentionally untouched; the
// lints below predate the move and are tracked as tech debt for a 0.9.x cleanup pass.
@@ -41,8 +39,6 @@ mod indexes_mut;
pub mod iobuf;
pub mod log;
mod memory_pool;
-mod messages_batch_mut;
-mod messages_batch_set;
mod segment_storage;
pub mod send_messages;
pub mod sharding;
@@ -55,11 +51,7 @@ pub use consensus_message::{
MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind,
};
pub use executor::create_shard_executor;
-pub use in_flight::IggyMessagesBatchSetInFlight;
-pub use indexes_mut::IggyIndexesMut;
pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolConfigOther, memory_pool};
-pub use messages_batch_mut::IggyMessagesBatchMut;
-pub use messages_batch_set::IggyMessagesBatchSet;
pub use segment_storage::{
IndexReader, IndexWriter, MessagesReader, MessagesWriter, SegmentStorage,
};
diff --git a/core/server_common/src/messages_batch_mut.rs b/core/server_common/src/messages_batch_mut.rs
deleted file mode 100644
index bf7bbd10e..000000000
--- a/core/server_common/src/messages_batch_mut.rs
+++ /dev/null
@@ -1,616 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use crate::PooledBuffer;
-use crate::indexes_mut::IggyIndexesMut;
-use iggy_common::{
- IGGY_MESSAGE_HEADER_SIZE, INDEX_SIZE, IggyByteSize, IggyError, IggyIndexView, IggyMessage,
- IggyMessageBoundaries, IggyMessageView, IggyMessageViewIterator, IggyMessageViewMutIterator,
- IggyMessagesBatch, MAX_PAYLOAD_SIZE, MAX_USER_HEADERS_SIZE, Sizeable, Validatable,
-};
-use std::ops::Index;
-use tracing::error;
-
-/// A container for mutable messages that are being prepared for persistence.
-///
-/// `IggyMessagesBatchMut` holds both the raw message data in a `BytesMut` buffer
-/// and the corresponding index data that allows for efficient message lookup.
-#[derive(Debug, Default)]
-pub struct IggyMessagesBatchMut {
- /// The index data for all messages in the buffer
- indexes: IggyIndexesMut,
-
- /// The buffer containing the serialized message data
- messages: PooledBuffer,
-}
-
-impl Sizeable for IggyMessagesBatchMut {
- fn get_size_bytes(&self) -> IggyByteSize {
- IggyByteSize::from(self.messages.len() as u64)
- }
-}
-
-impl IggyMessagesBatchMut {
- /// Creates a new empty messages container
- pub fn empty() -> Self {
- Self {
- indexes: IggyIndexesMut::empty(),
- messages: PooledBuffer::empty(),
- }
- }
-
- /// Creates a new messages container from existing index and message buffers.
- ///
- /// # Arguments
- ///
- /// * `indexes` - Preprocessed index data
- /// * `messages` - Serialized message data
- pub fn from_indexes_and_messages(indexes: IggyIndexesMut, messages: PooledBuffer) -> Self {
- Self { indexes, messages }
- }
-
- /// Creates a new messages container from a slice of IggyMessage objects.
- ///
- /// # Note
- /// This function should be used only for testing purposes,
- /// because it creates deep copies of the messages.
- ///
- /// # Arguments
- ///
- /// * `messages` - Slice of message objects to store
- /// * `messages_size` - Total size of all messages in bytes
- pub fn from_messages(messages: &[IggyMessage], messages_size: u32) -> Self {
- let mut messages_buffer = PooledBuffer::with_capacity(messages_size as usize);
- let mut indexes_buffer = IggyIndexesMut::with_capacity(messages.len(), 0);
- let mut position = 0;
-
- for message in messages {
- let bytes = message.to_bytes();
- messages_buffer.put_slice(&bytes);
- position += message.get_size_bytes().as_bytes_u32();
- indexes_buffer.insert(0, position, 0);
- }
-
- Self::from_indexes_and_messages(indexes_buffer, messages_buffer)
- }
-
- /// Creates a lending iterator that yields mutable views of messages.
- pub fn iter_mut(&mut self) -> IggyMessageViewMutIterator<'_> {
- IggyMessageViewMutIterator::new(&mut self.messages)
- }
-
- /// Creates an iterator that yields immutable views of messages.
- pub fn iter(&self) -> IggyMessageViewIterator<'_> {
- IggyMessageViewIterator::new_with_boundaries(
- &self.messages,
- &self.indexes,
- self.indexes.base_position(),
- self.count(),
- )
- }
-
- /// Returns the number of messages in the batch.
- pub fn count(&self) -> u32 {
- self.indexes.len() as u32 / INDEX_SIZE as u32
- }
-
- /// Returns the total size of all messages in bytes.
- pub fn size(&self) -> u32 {
- self.messages.len() as u32
- }
-
- /// Returns the raw message bytes as a slice.
- pub fn as_bytes(&self) -> &[u8] {
- &self.messages
- }
-
- /// Returns the first offset in the batch
- pub fn first_offset(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
- Some(IggyMessageView::new(&self.messages).ok()?.header().offset())
- }
-
- /// Returns the first timestamp in the batch
- pub fn first_timestamp(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
- Some(
- IggyMessageView::new(&self.messages)
- .ok()?
- .header()
- .timestamp(),
- )
- }
-
- /// Returns the last timestamp in the batch
- pub fn last_timestamp(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
-
- let last_index = self.count() as usize - 1;
- let (start, end) = self.get_message_boundaries(last_index)?;
- Some(
- IggyMessageView::new(&self.messages[start..end])
- .ok()?
- .header()
- .timestamp(),
- )
- }
-
- /// Returns the last offset in the batch
- pub fn last_offset(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
- let last_index = self.count() as usize - 1;
- let (start, end) = self.get_message_boundaries(last_index)?;
- Some(
- IggyMessageView::new(&self.messages[start..end])
- .ok()?
- .header()
- .offset(),
- )
- }
-
- /// Checks if the batch is empty.
- pub fn is_empty(&self) -> bool {
- self.count() == 0
- }
-
- /// Return total size of all messages in bytes.
- pub fn len(&self) -> usize {
- self.messages.len()
- }
-
- /// Decomposes the batch into its constituent parts.
- pub fn decompose(mut self) -> (IggyIndexesMut, PooledBuffer) {
- let indexes = std::mem::replace(&mut self.indexes, IggyIndexesMut::empty());
- let messages = std::mem::take(&mut self.messages);
-
- (indexes, messages)
- }
-
- /// Freezes the batch, converting to an immutable `IggyMessagesBatch`.
- ///
- /// The returned batch uses Arc-backed `Bytes`, allowing cheap clones.
- /// After calling this, the mutable batch becomes empty.
- pub fn freeze(&mut self) -> IggyMessagesBatch {
- let count = self.count();
- let indexes = self.indexes.freeze();
- let messages = self.messages.freeze();
- IggyMessagesBatch::new(indexes, messages, count)
- }
-
- pub fn take_messages(&mut self) -> PooledBuffer {
- std::mem::take(&mut self.messages)
- }
-
- /// Take the indexes from the batch
- pub fn take_indexes(&mut self) -> IggyIndexesMut {
- std::mem::take(&mut self.indexes)
- }
-
- /// Borrows the indexes from the batch
- pub fn indexes(&self) -> &IggyIndexesMut {
- &self.indexes
- }
-
- /// Get message position from the indexes at the given index
- pub fn position_at(&self, index: u32) -> Option<u32> {
- self.indexes.get(index).map(|index| index.position())
- }
-
- fn boundaries(&self) -> Option<IggyMessageBoundaries<'_>> {
- IggyMessageBoundaries::new(
- &self.indexes,
- self.messages.len(),
- self.indexes.base_position(),
- self.count(),
- )
- }
-
- /// Calculates the start position of a message at the given index in the buffer
- fn message_start_position(&self, index: usize) -> Option<usize> {
- self.get_message_boundaries(index).map(|(start, _)| start)
- }
-
- /// Calculates the end position of a message at the given index in the buffer
- fn message_end_position(&self, index: usize) -> Option<usize> {
- self.get_message_boundaries(index).map(|(_, end)| end)
- }
-
- /// Returns a contiguous slice (as a new `IggyMessagesBatch`) of up to `count` messages
- /// whose message headers have an offset greater than or equal to the provided `start_offset`.
- pub fn slice_by_offset(&self, start_offset: u64, count: u32) -> Option<Self> {
- if self.is_empty() || count == 0 {
- return None;
- }
-
- let first_offset = self.first_offset()?;
-
- if start_offset < first_offset {
- return None;
- }
-
- let last_offset = self.last_offset()?;
- if start_offset > last_offset {
- return None;
- }
-
- let offset_diff = start_offset - first_offset;
- let first_message_index = offset_diff as usize;
-
- if first_message_index >= self.count() as usize {
- return None;
- }
-
- self.slice_by_index(first_message_index as u32, count)
- }
-
- /// Helper method to slice the batch starting from a specific index
- fn slice_by_index(&self, start_index: u32, count: u32) -> Option<Self> {
- if start_index >= self.count() {
- return None;
- }
-
- let last_message_index =
- std::cmp::min((start_index + count) as usize, self.count() as usize);
-
- let sub_indexes = self.indexes.slice_by_offset(
- start_index,
- (last_message_index - start_index as usize) as u32,
- )?;
-
- let first_message_position = self.message_start_position(start_index as usize)?;
- let last_message_position = self.message_end_position(last_message_index - 1)?;
-
- // TODO(hubcio): messages from accumulator unfortunately are deep-copied
- let mut sub_buffer =
- PooledBuffer::with_capacity(last_message_position - first_message_position);
- sub_buffer.put_slice(&self.messages[first_message_position..last_message_position]);
-
- Some(IggyMessagesBatchMut {
- indexes: sub_indexes,
- messages: sub_buffer,
- })
- }
-
- /// Returns a contiguous slice (as a new `IggyMessagesBatch`) of up to `count` messages
- /// whose message headers have a timestamp greater than or equal to the provided `timestamp`.
- ///
- /// If no messages meet the criteria, returns `None`.
- pub fn slice_by_timestamp(&self, timestamp: u64, count: u32) -> Option<Self> {
- if self.is_empty() || count == 0 {
- return None;
- }
-
- // Use binary search to find the first message with timestamp >= the target
- let first_message_index = self.binary_search_timestamp(timestamp)?;
-
- self.slice_by_index(first_message_index, count)
- }
-
- /// Find the position of the index with timestamp closest to (but not exceeding) the target
- fn binary_search_timestamp(&self, target_timestamp: u64) -> Option<u32> {
- if self.count() == 0 {
- return None;
- }
-
- let last_timestamp = self.get(self.count() as usize - 1)?.header().timestamp();
- if target_timestamp > last_timestamp {
- return Some(self.count() - 1);
- }
-
- let first_timestamp = self.get(0)?.header().timestamp();
- if target_timestamp <= first_timestamp {
- return Some(0);
- }
-
- let mut low = 0;
- let mut high = self.count() - 1;
-
- while low <= high {
- let mid = low + (high - low) / 2;
- let mid_index = self.get(mid as usize)?;
- let mid_timestamp = mid_index.header().timestamp();
-
- match mid_timestamp.cmp(&target_timestamp) {
- std::cmp::Ordering::Equal => return Some(mid),
- std::cmp::Ordering::Less => low = mid + 1,
- std::cmp::Ordering::Greater => {
- if mid == 0 {
- break;
- }
- high = mid - 1;
- }
- }
- }
-
- Some(low)
- }
-
- /// Validates that all messages have correct checksums and offsets.
- /// This function should be called after messages have been read from disk.
- ///
- /// # Arguments
- ///
- /// * `absolute_start_offset` - The absolute offset of the first message in the batch.
- ///
- /// # Returns
- ///
- /// * `Ok(())` - If all messages have correct checksums and offsets.
- /// * `Err(IggyError)` - If any message has an invalid checksum or offset.
- pub fn validate_checksums_and_offsets(
- &self,
- absolute_start_offset: u64,
- ) -> Result<(), IggyError> {
- for (current_offset, message) in (absolute_start_offset..).zip(self.iter()) {
- let calculated_checksum = message.calculate_checksum();
- let actual_checksum = message.header().checksum();
- let offset = message.header().offset();
- if offset != current_offset {
- return Err(IggyError::InvalidOffset(offset));
- }
- if calculated_checksum != actual_checksum {
- return Err(IggyError::InvalidMessageChecksum(
- actual_checksum,
- calculated_checksum,
- offset,
- ));
- }
- }
- Ok(())
- }
-
- /// Gets the byte range for a message at the given index
- fn get_message_boundaries(&self, index: usize) -> Option<(usize, usize)> {
- let (start, end) = self.boundaries()?.boundaries(index)?;
-
- if start > self.messages.len()
- || end > self.messages.len()
- || start > end
- || end - start < IGGY_MESSAGE_HEADER_SIZE
- {
- return None;
- }
-
- Some((start, end))
- }
-
- /// Get the message at the specified index.
- /// Returns None if the index is out of bounds or the message cannot be found.
- pub fn get(&self, index: usize) -> Option<IggyMessageView<'_>> {
- let (start, end) = self.get_message_boundaries(index)?;
- IggyMessageView::new(&self.messages[start..end]).ok()
- }
-
- /// Validates that all messages in batch have correct checksums.
- pub fn validate_checksums(&self) -> Result<(), IggyError> {
- for message in self.iter() {
- let calculated_checksum = message.calculate_checksum();
- let actual_checksum = message.header().checksum();
- let offset = message.header().offset();
- if calculated_checksum != actual_checksum {
- return Err(IggyError::InvalidMessageChecksum(
- actual_checksum,
- calculated_checksum,
- offset,
- ));
- }
- }
- Ok(())
- }
-
- /// Validates the structure of the indexes (sizes, counts, etc.)
- fn validate_indexes_structure(&self) -> Result<(), IggyError> {
- let indexes_count = self.indexes.count();
- let indexes_size = self.indexes.size();
-
- if !indexes_size.is_multiple_of(INDEX_SIZE as u32) {
- error!(
- "Indexes size {} is not a multiple of index size {}",
- indexes_size, INDEX_SIZE
- );
- return Err(IggyError::InvalidIndexesByteSize(indexes_size));
- }
-
- if indexes_count != self.count() {
- error!(
- "Indexes count {} does not match messages count {}",
- indexes_count,
- self.count()
- );
- return Err(IggyError::InvalidIndexesCount(indexes_count, self.count()));
- }
-
- Ok(())
- }
-
- fn validate_message_contents(&self) -> Result<(), IggyError> {
- let mut messages_count = 0;
- let mut messages_size = 0;
- let mut prev_offset = 0;
- let mut prev_position = 0;
-
- for i in 0..self.count() {
- let index = self.validate_index_at(i)?;
- let message = self.validate_message_at(i)?;
-
- if message.header().offset() < prev_offset {
- error!(
- "Offset of previous message: {} is smaller than current message {} at offset {}",
- prev_offset,
- message.header().offset(),
- i
- );
- return Err(IggyError::InvalidOffset(message.header().offset()));
- }
-
- if index.position() < prev_position {
- error!(
- "Position of previous message: {} is smaller than current message {} at offset {}",
- prev_position,
- index.position(),
- i
- );
- return Err(IggyError::CannotReadIndexPosition);
- }
-
- prev_offset = message.header().offset();
- prev_position = index.position();
- messages_size += message.size();
- messages_count += 1;
- }
-
- let indexes_count = self.indexes.count();
- if indexes_count != messages_count {
- error!(
- "Indexes count {} does not match messages count {}",
- indexes_count, messages_count
- );
- return Err(IggyError::InvalidMessagesCount);
- }
-
- if messages_size != self.messages.len() {
- error!(
- "Messages size {} does not match messages buffer size {}",
- messages_size,
- self.messages.len() as u64
- );
- return Err(IggyError::InvalidMessagesSize(
- messages_size as u32,
- self.messages.len() as u32,
- ));
- }
-
- Ok(())
- }
-
- /// Validates a specific index entry
- fn validate_index_at(&self, i: u32) -> Result<IggyIndexView<'_>, IggyError> {
- let index_view = match self.indexes.get(i) {
- Some(view) => view,
- None => {
- error!("Index {} is missing", i);
- return Err(IggyError::MissingIndex(i));
- }
- };
-
- if index_view.offset() != 0 {
- error!(
- "Non-zero offset {} at index: {}, messages coming from network always have offset 0",
- index_view.offset(),
- i
- );
- return Err(IggyError::NonZeroOffset(index_view.offset() as u64, i));
- }
-
- if index_view.timestamp() != 0 {
- error!(
- "Non-zero timestamp {} at index: {}, messages coming from network always have timestamp 0",
- index_view.timestamp(),
- i
- );
- return Err(IggyError::NonZeroTimestamp(index_view.timestamp(), i));
- }
-
- Ok(index_view)
- }
-
- /// Validates a specific message
- fn validate_message_at(&self, i: u32) -> Result<IggyMessageView<'_>, IggyError> {
- let message = match self.get(i as usize) {
- Some(msg) => msg,
- None => {
- error!(
- "Message at index {} is missing, or message size is less than minimum message size {} B (header)",
- i, IGGY_MESSAGE_HEADER_SIZE
- );
- return Err(IggyError::MissingIndex(i));
- }
- };
-
- if message.payload().len() as u32 > MAX_PAYLOAD_SIZE {
- error!(
- "Message payload size {} B exceeds maximum payload size {} B",
- message.payload().len(),
- MAX_PAYLOAD_SIZE
- );
- return Err(IggyError::TooBigMessagePayload);
- }
-
- if message.size() < IGGY_MESSAGE_HEADER_SIZE {
- error!(
- "Message size {} B is less than minimum message size {} B (header)",
- message.size(),
- IGGY_MESSAGE_HEADER_SIZE
- );
- return Err(IggyError::TooSmallMessage(
- message.size() as u32,
- IGGY_MESSAGE_HEADER_SIZE as u32,
- ));
- }
-
- if let Some(user_headers) = message.user_headers()
- && user_headers.len() as u32 > MAX_USER_HEADERS_SIZE
- {
- error!(
- "Message user headers size {} B exceeds maximum size {} B",
- user_headers.len(),
- MAX_USER_HEADERS_SIZE
- );
- return Err(IggyError::TooBigUserHeaders);
- }
-
- Ok(message)
- }
-}
-
-impl Validatable<IggyError> for IggyMessagesBatchMut {
- fn validate(&self) -> Result<(), IggyError> {
- if self.is_empty() {
- return Err(IggyError::InvalidMessagesCount);
- }
-
- self.validate_indexes_structure()?;
- self.validate_message_contents()?;
- Ok(())
- }
-}
-
-impl Index<usize> for IggyMessagesBatchMut {
- type Output = [u8];
-
- fn index(&self, index: usize) -> &Self::Output {
- assert!(
- index < self.count() as usize,
- "Index out of bounds: the len is {} but the index is {}",
- self.count(),
- index
- );
-
- let (start, end) = self
- .get_message_boundaries(index)
- .expect("Invalid message boundaries");
-
- &self.messages[start..end]
- }
-}
diff --git a/core/server_common/src/messages_batch_set.rs b/core/server_common/src/messages_batch_set.rs
deleted file mode 100644
index a10bda3ea..000000000
--- a/core/server_common/src/messages_batch_set.rs
+++ /dev/null
@@ -1,378 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use crate::{IggyIndexesMut, IggyMessagesBatchMut, PooledBuffer};
-use bytes::Bytes;
-use iggy_common::{
- IggyByteSize, IggyMessage, IggyMessageView, IggyMessagesBatch, IggyPollMetadata,
- PolledMessages, Sizeable,
-};
-use std::ops::Index;
-use tracing::trace;
-
-/// A container for multiple IggyMessagesBatch objects
-#[derive(Debug, Default)]
-pub struct IggyMessagesBatchSet {
- /// The collection of message containers
- batches: Vec<IggyMessagesBatchMut>,
- /// Total number of messages across all containers
- count: u32,
- /// Total size in bytes across all containers
- size: u32,
-}
-
-impl IggyMessagesBatchSet {
- /// Create a new empty batch
- pub fn empty() -> Self {
- Self {
- batches: Vec::new(),
- count: 0,
- size: 0,
- }
- }
-
- /// Create a new empty batch set with a specified initial capacity of message containers
- pub fn with_capacity(capacity: usize) -> Self {
- Self {
- batches: Vec::with_capacity(capacity),
- count: 0,
- size: 0,
- }
- }
-
- /// Create a batch set from an existing vector of IggyMessages
- pub fn from_vec(messages: Vec<IggyMessagesBatchMut>) -> Self {
- let mut batch = Self::with_capacity(messages.len());
- for msg in messages {
- batch.add_batch(msg);
- }
- batch
- }
-
- /// Add another batch of messages to the batch set
- pub fn add_batch(&mut self, batch: IggyMessagesBatchMut) {
- self.count += batch.count();
- self.size += batch.size();
- self.batches.push(batch);
- }
-
- /// Add another batch set of messages to the batch set
- pub fn add_batch_set(&mut self, mut other_batch_set: IggyMessagesBatchSet) {
- self.count += other_batch_set.count();
- self.size += other_batch_set.size();
- let other_batches = std::mem::take(&mut other_batch_set.batches);
- self.batches.extend(other_batches);
- }
-
- /// Add immutable batches by copying them into mutable form.
- ///
- /// This is used when reading from the in-flight buffer (which holds
- /// frozen/immutable batches) and needs to convert them for the read path.
- pub fn add_immutable_batches(&mut self, batches: &[IggyMessagesBatch]) {
- for batch in batches {
- let mutable_batch = Self::immutable_to_mutable(batch);
- self.add_batch(mutable_batch);
- }
- }
-
- /// Convert an immutable IggyMessagesBatch to a mutable IggyMessagesBatchMut.
- ///
- /// This requires copying the Bytes into a PooledBuffer.
- fn immutable_to_mutable(batch: &IggyMessagesBatch) -> IggyMessagesBatchMut {
- let base_position = batch.indexes().base_position();
- let indexes_buffer = PooledBuffer::from(batch.indexes_slice());
- let indexes = IggyIndexesMut::from_bytes(indexes_buffer, base_position);
- let messages = PooledBuffer::from(batch.buffer());
-
- IggyMessagesBatchMut::from_indexes_and_messages(indexes, messages)
- }
-
- /// Extract indexes from all batches in the set
- pub fn append_indexes_to(&self, target: &mut IggyIndexesMut) {
- for batch in self.iter() {
- let indexes = batch.indexes();
- target.append_slice(indexes);
- }
- }
-
- /// Get the total number of messages in the batch
- pub fn count(&self) -> u32 {
- self.count
- }
-
- /// Get the total size of all messages in bytes
- pub fn size(&self) -> u32 {
- self.size
- }
-
- /// Get the number of message containers in the batch
- pub fn containers_count(&self) -> usize {
- self.batches.len()
- }
-
- /// Check if the batch is empty
- pub fn is_empty(&self) -> bool {
- self.batches.is_empty() || self.count == 0
- }
-
- /// Get timestamp of first message in first batch
- pub fn first_timestamp(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
- self.batches.first().map(|batch| batch.first_timestamp())?
- }
-
- /// Get offset of first message in first batch
- pub fn first_offset(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
- self.batches.first().map(|batch| batch.first_offset())?
- }
-
- /// Get timestamp of last message in last batch
- pub fn last_timestamp(&self) -> Option<u64> {
- if self.is_empty() {
- return None;
- }
- self.batches.last().map(|batch| batch.last_timestamp())?
- }
-
- /// Get offset of last message in last batch
- pub fn last_offset(&self) -> Option<u64> {
- self.batches.last().map(|batch| batch.last_offset())?
- }
-
- /// Get a reference to the underlying vector of message containers
- pub fn inner(&self) -> &Vec<IggyMessagesBatchMut> {
- &self.batches
- }
-
- /// Consume the batch, returning the underlying vector of message containers
- pub fn into_inner(mut self) -> Vec<IggyMessagesBatchMut> {
- std::mem::take(&mut self.batches)
- }
-
- /// Iterate over all message containers in the batch
- pub fn iter(&self) -> impl Iterator<Item = &IggyMessagesBatchMut> {
- self.batches.iter()
- }
-
- /// Iterate over all mutable message containers in the batch
- pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut IggyMessagesBatchMut> {
- self.batches.iter_mut()
- }
-
- /// Convert this batch and poll metadata into a vector of fully-formed IggyMessage objects
- ///
- /// This method transforms the internal message views into complete IggyMessage objects
- /// that can be returned to clients. It should only be used by server http implementation.
- ///
- /// # Arguments
- ///
- /// * `poll_metadata` - Metadata about the partition and current offset
- ///
- /// # Returns
- ///
- /// A vector of IggyMessage objects with proper metadata
- pub fn into_polled_messages(&self, poll_metadata: IggyPollMetadata) -> PolledMessages {
- if self.is_empty() {
- return PolledMessages::empty();
- }
-
- let mut messages = Vec::with_capacity(self.count() as usize);
-
- for batch in self.iter() {
- for message in batch.iter() {
- let header = message.header().to_header();
- let payload = Bytes::copy_from_slice(message.payload());
- let user_headers = message.user_headers().map(Bytes::copy_from_slice);
- let message = IggyMessage {
- header,
- payload,
- user_headers,
- };
- messages.push(message);
- }
- }
-
- trace!(
- "Converted batch of {} messages from partition {} with current offset {}",
- messages.len(),
- poll_metadata.partition_id,
- poll_metadata.current_offset
- );
-
- PolledMessages {
- partition_id: poll_metadata.partition_id,
- current_offset: poll_metadata.current_offset,
- count: messages.len() as u32,
- messages,
- }
- }
-
- /// Returns a new IggyMessagesBatch containing only messages with offsets greater than or equal to the specified offset,
- /// up to the specified count.
- ///
- /// If no messages match the criteria, returns an empty batch.
- pub fn get_by_offset(&self, start_offset: u64, count: u32) -> Self {
- if self.is_empty() || count == 0 {
- return Self::empty();
- }
-
- let mut result = Self::with_capacity(self.containers_count());
- let mut remaining_count = count;
- let mut current_offset = start_offset;
-
- for container in self.iter() {
- if remaining_count == 0 {
- break;
- }
-
- let Some(batch_first) = container.first_offset() else {
- continue;
- };
- if batch_first + container.count() as u64 <= current_offset {
- continue;
- }
-
- // When current_offset is below this batch's range (cross-batch
- // reads), start from the batch's first offset instead.
- let effective_start = current_offset.max(batch_first);
-
- if let Some(sliced) = container.slice_by_offset(effective_start, remaining_count)
- && sliced.count() > 0
- {
- if let Some(last) = sliced.last_offset() {
- current_offset = last + 1;
- }
- remaining_count -= sliced.count();
- result.add_batch(sliced);
- }
- }
-
- result
- }
-
- /// Returns a new IggyMessagesBatch containing only messages with timestamps greater than or equal
- /// to the specified timestamp, up to the specified count.
- ///
- /// If no messages match the criteria, returns an empty batch.
- pub fn get_by_timestamp(&self, timestamp: u64, count: u32) -> Self {
- if self.is_empty() || count == 0 {
- return Self::empty();
- }
-
- let mut result = Self::with_capacity(self.containers_count());
- let mut remaining_count = count;
-
- for container in self.iter() {
- if remaining_count == 0 {
- break;
- }
-
- let first_timestamp = container.first_timestamp();
- if first_timestamp.is_none() || first_timestamp.unwrap() < timestamp {
- continue;
- }
-
- if let Some(sliced) = container.slice_by_timestamp(timestamp, remaining_count)
- && sliced.count() > 0
- {
- remaining_count -= sliced.count();
- result.add_batch(sliced);
- }
- }
-
- result
- }
-
- /// Get the message at the specified index.
- /// Returns None if the index is out of bounds.
- pub fn get(&self, index: usize) -> Option<IggyMessageView<'_>> {
- if index >= self.count as usize {
- return None;
- }
-
- let mut seen_messages = 0;
-
- for batch in &self.batches {
- let batch_count = batch.count() as usize;
-
- if index < seen_messages + batch_count {
- let local_index = index - seen_messages;
- return batch.get(local_index);
- }
-
- seen_messages += batch_count;
- }
-
- None
- }
-}
-
-impl Index<usize> for IggyMessagesBatchSet {
- type Output = [u8];
-
- /// Get the message bytes at the specified index across all batches
- ///
- /// # Panics
- ///
- /// Panics if the index is out of bounds (>= total number of messages)
- fn index(&self, index: usize) -> &Self::Output {
- assert!(
- index < self.count as usize,
- "Index out of bounds: the len is {} but the index is {}",
- self.count,
- index
- );
-
- let mut seen_messages = 0;
-
- for batch in &self.batches {
- let batch_count = batch.count() as usize;
-
- if index < seen_messages + batch_count {
- let local_index = index - seen_messages;
- return &batch[local_index];
- }
-
- seen_messages += batch_count;
- }
-
- unreachable!("Failed to find message at index {}", index);
- }
-}
-
-impl Sizeable for IggyMessagesBatchSet {
- fn get_size_bytes(&self) -> IggyByteSize {
- IggyByteSize::from(self.size as u64)
- }
-}
-
-impl From<Vec<IggyMessagesBatchMut>> for IggyMessagesBatchSet {
- fn from(messages: Vec<IggyMessagesBatchMut>) -> Self {
- Self::from_vec(messages)
- }
-}
-
-impl From<IggyMessagesBatchMut> for IggyMessagesBatchSet {
- fn from(messages: IggyMessagesBatchMut) -> Self {
- Self::from_vec(vec![messages])
- }
-}
diff --git a/core/server_common/src/segment_storage/index_reader.rs b/core/server_common/src/segment_storage/index_reader.rs
index 7f45c840b..b4fb65140 100644
--- a/core/server_common/src/segment_storage/index_reader.rs
+++ b/core/server_common/src/segment_storage/index_reader.rs
@@ -15,375 +15,37 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{IggyIndexesMut, PooledBuffer};
-use compio::{
- buf::{IntoInner, IoBuf},
- fs::{File, OpenOptions},
- io::AsyncReadAtExt,
-};
+use compio::fs::OpenOptions;
use err_trail::ErrContext;
-use iggy_common::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView};
-use std::{
- io::ErrorKind,
- rc::Rc,
- sync::atomic::{AtomicU64, Ordering},
-};
-use tracing::{error, trace};
+use iggy_common::IggyError;
+use tracing::trace;
-/// A dedicated struct for reading from the index file.
+/// Path handle for a segment's index file, validated openable at segment
+/// build. Reads go through the partition's own index reader; this exists so
+/// storage plumbing (bootstrap, state transfer) can resolve the index path.
#[derive(Debug)]
pub struct IndexReader {
file_path: String,
- file: File,
- index_size_bytes: Rc<AtomicU64>,
}
-// Safety: We are guaranteeing that IndexReader will never be used from multiple threads
-unsafe impl Send for IndexReader {}
-
impl IndexReader {
- /// Opens the index file in read-only mode.
- pub async fn new(file_path: &str, index_size_bytes: Rc<AtomicU64>) -> Result<Self, IggyError> {
- let file = OpenOptions::new()
+ /// Opens the index file read-only to prove it exists, then drops the
+ /// descriptor: nothing reads through this type.
+ pub async fn new(file_path: &str) -> Result<Self, IggyError> {
+ OpenOptions::new()
.read(true)
.open(file_path)
.await
.error(|e: &std::io::Error| format!("Failed to open index file: {file_path}. {e}"))
.map_err(|_| IggyError::CannotReadFile)?;
- trace!(
- "Opened index file for reading: {file_path}, size: {}",
- index_size_bytes.load(Ordering::Acquire)
- );
+ trace!("Validated index file for reading: {file_path}");
Ok(Self {
file_path: file_path.to_string(),
- file,
- index_size_bytes,
})
}
pub fn path(&self) -> String {
self.file_path.clone()
}
-
- /// Loads all indexes from the index file into the optimized binary format.
- /// Note that this function does not use the pool, as the messages are not cached.
- /// This is expected - this method is called at startup and we want to preserve
- /// memory pool usage.
- pub async fn load_all_indexes_from_disk(&self) -> Result<IggyIndexesMut, IggyError> {
- let file_size = self.file_size();
- if file_size == 0 {
- return Ok(IggyIndexesMut::empty());
- }
-
- let buf = match self.read_at(0, file_size, false).await {
- Ok(buf) => buf,
- Err(e) if e.kind() == ErrorKind::UnexpectedEof => {
- return Ok(IggyIndexesMut::empty());
- }
- Err(e) => {
- error!(
- "Error reading batch header at offset 0 in file {}: {e}",
- self.file_path
- );
- return Err(IggyError::CannotReadFile);
- }
- };
- let index_count = file_size / INDEX_SIZE as u32;
- let mut indexes = IggyIndexesMut::from_bytes(buf, 0);
- if indexes.count() != index_count {
- error!(
- "Loaded {} indexes from disk, expected {}, file {} is probably corrupted!",
- indexes.count(),
- index_count,
- self.file_path
- );
- }
- indexes.mark_saved();
- Ok(indexes)
- }
-
- /// Loads a specific range of indexes from disk based on offset.
- ///
- /// Returns a slice of indexes starting at the relative_start_offset with the specified count,
- /// or None if the requested range is not available.
- pub async fn load_from_disk_by_offset(
- &self,
- relative_start_offset: u32,
- count: u32,
- ) -> Result<Option<IggyIndexesMut>, IggyError> {
- let file_size = self.file_size();
- let total_indexes = file_size / INDEX_SIZE as u32;
-
- if file_size == 0 || total_indexes == 0 {
- trace!(
- "Index file {} is empty, cannot load indexes",
- self.file_path
- );
- return Ok(None);
- }
-
- if relative_start_offset >= total_indexes {
- trace!(
- "Start offset {} is out of bounds. Total indexes: {}",
- relative_start_offset, total_indexes
- );
- return Ok(None);
- }
-
- let available_count = total_indexes.saturating_sub(relative_start_offset);
- let actual_count = std::cmp::min(count, available_count);
-
- if actual_count == 0 {
- trace!(
- "No available indexes to load. Start offset: {}, requested count: {}",
- relative_start_offset, count
- );
- return Ok(None);
- }
-
- let start_byte = relative_start_offset as usize * INDEX_SIZE;
- let end_byte = start_byte + (actual_count as usize * INDEX_SIZE);
-
- let indexes_bytes = match self
- .read_at(start_byte as u32, (end_byte - start_byte) as u32, true)
- .await
- {
- Ok(buf) => buf,
- Err(e) if e.kind() == ErrorKind::UnexpectedEof => {
- error!("Unexpected EOF while reading indexes");
- return Ok(None);
- }
- Err(e) => {
- error!(
- "Error reading {actual_count} indexes at position {relative_start_offset} in file {} of size {file_size}: {e}",
- self.file_path
- );
- return Err(IggyError::CannotReadFile);
- }
- };
-
- let base_position = if relative_start_offset > 0 {
- match self.load_nth_index(relative_start_offset - 1).await? {
- Some(prev_index) => prev_index.position,
- None => {
- trace!(
- "Failed to load previous index at position {}",
- relative_start_offset - 1
- );
- 0
- }
- }
- } else {
- 0
- };
-
- let indexes = IggyIndexesMut::from_bytes(indexes_bytes, base_position);
-
- trace!(
- "Loaded {actual_count} indexes from disk starting at offset {relative_start_offset}, base position: {base_position}, last position: {}",
- indexes.last_position()
- );
-
- Ok(Some(indexes))
- }
-
- /// Loads a specific range of indexes from disk based on timestamp.
- ///
- /// Returns a slice of indexes starting from the index with timestamp closest to
- /// (but not exceeding) the requested timestamp, with the specified count.
- pub async fn load_from_disk_by_timestamp(
- &self,
- timestamp: u64,
- count: u32,
- ) -> Result<Option<IggyIndexesMut>, IggyError> {
- let file_size = self.file_size();
- let total_indexes = file_size / INDEX_SIZE as u32;
-
- if file_size == 0 || total_indexes == 0 {
- trace!("Index file is empty");
- return Ok(None);
- }
-
- let start_index_pos = match self
- .binary_search_position_for_timestamp_async(timestamp)
- .await?
- {
- Some(pos) => pos,
- None => return Ok(None),
- };
-
- let available_count = total_indexes.saturating_sub(start_index_pos);
- let actual_count = std::cmp::min(count, available_count);
-
- if actual_count == 0 {
- trace!(
- "No available indexes to load. Start index pos: {}, requested count: {}",
- start_index_pos, count
- );
- return Ok(None);
- }
-
- let start_byte = start_index_pos as usize * INDEX_SIZE;
- let end_byte = start_byte + (actual_count as usize * INDEX_SIZE);
-
- let indexes_bytes = match self
- .read_at(start_byte as u32, (end_byte - start_byte) as u32, true)
- .await
- {
- Ok(buf) => buf,
- Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(None),
- Err(e) => {
- error!(
- "Error reading {actual_count} indexes at position {start_index_pos} in file {}: {e}",
- self.file_path
- );
- return Err(IggyError::CannotReadFile);
- }
- };
-
- let base_position = if start_index_pos > 0 {
- match self.load_nth_index(start_index_pos - 1).await? {
- Some(prev_index) => prev_index.position,
- None => {
- trace!(
- "Failed to load previous index at position {}",
- start_index_pos - 1
- );
- 0
- }
- }
- } else {
- 0
- };
-
- trace!(
- "Loaded {} indexes from disk starting at timestamp {}, base position: {}",
- actual_count, timestamp, base_position
- );
-
- Ok(Some(IggyIndexesMut::from_bytes(
- indexes_bytes,
- base_position,
- )))
- }
-
- /// Finds the position of the index with timestamp closest to (but not exceeding) the target
- async fn binary_search_position_for_timestamp_async(
- &self,
- target_timestamp: u64,
- ) -> Result<Option<u32>, IggyError> {
- let file_size = self.file_size();
- if file_size == 0 {
- return Ok(None);
- }
-
- let total_indexes = file_size / INDEX_SIZE as u32;
- if total_indexes == 0 {
- return Ok(None);
- }
-
- let last_idx = match self.load_nth_index(total_indexes - 1).await? {
- Some(idx) => idx,
- None => return Ok(None),
- };
-
- if target_timestamp > last_idx.timestamp {
- return Ok(Some(total_indexes - 1));
- }
-
- let first_idx = match self.load_nth_index(0).await? {
- Some(idx) => idx,
- None => return Ok(None),
- };
-
- if target_timestamp <= first_idx.timestamp {
- return Ok(Some(0));
- }
-
- let mut low = 0;
- let mut high = total_indexes - 1;
-
- while low <= high {
- let mid = low + (high - low) / 2;
- let mid_index = match self.load_nth_index(mid).await? {
- Some(idx) => idx,
- None => break,
- };
-
- match mid_index.timestamp.cmp(&target_timestamp) {
- std::cmp::Ordering::Equal => return Ok(Some(mid)),
- std::cmp::Ordering::Less => low = mid + 1,
- std::cmp::Ordering::Greater => {
- if mid == 0 {
- break;
- }
- high = mid - 1;
- }
- }
- }
-
- Ok(Some(low))
- }
-
- /// Returns the size of the index file in bytes.
- fn file_size(&self) -> u32 {
- self.index_size_bytes.load(Ordering::Acquire) as u32
- }
-
- /// Reads a specified number of bytes from the index file at a given offset.
- async fn read_at(
- &self,
- offset: u32,
- len: u32,
- _use_pool: bool,
- ) -> Result<PooledBuffer, std::io::Error> {
- let buf = PooledBuffer::with_capacity(len as usize);
-
- let (result, buf) = self
- .file
- .read_exact_at(buf.slice(..len as usize), offset as u64)
- .await
- .into();
- let buf = buf.into_inner();
- result?;
- Ok(buf)
- }
-
- /// Gets the nth index from the index file.
- ///
- /// The index position is 0-based (first index is at position 0).
- /// Returns None if the specified position is out of bounds.
- async fn load_nth_index(&self, position: u32) -> Result<Option<IggyIndex>, IggyError> {
- let file_size = self.file_size();
- let total_indexes = file_size / INDEX_SIZE as u32;
-
- if position >= total_indexes {
- trace!(
- "Index position {} is out of bounds. Total indexes: {}",
- position, total_indexes
- );
- return Ok(None);
- }
-
- let offset = position * INDEX_SIZE as u32;
-
- let buf = match self.read_at(offset, INDEX_SIZE as u32, true).await {
- Ok(buf) => buf,
- Err(e) if e.kind() == ErrorKind::UnexpectedEof => {
- return Ok(None);
- }
- Err(e) => {
- error!(
- "Error reading index at position {} (offset {}) in file {}: {e}",
- position, offset, self.file_path
- );
- return Err(IggyError::CannotReadFile);
- }
- };
-
- let index = IggyIndexView::new(&buf).to_index();
-
- Ok(Some(index))
- }
}
diff --git a/core/server_common/src/segment_storage/index_writer.rs b/core/server_common/src/segment_storage/index_writer.rs
index 0ea1ce86c..f0603fb8a 100644
--- a/core/server_common/src/segment_storage/index_writer.rs
+++ b/core/server_common/src/segment_storage/index_writer.rs
@@ -15,12 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::PooledBuffer;
use compio::fs::File;
use compio::fs::OpenOptions;
-use compio::io::AsyncWriteAtExt;
use err_trail::ErrContext;
-use iggy_common::INDEX_SIZE;
use iggy_common::IggyError;
use std::rc::Rc;
use std::sync::atomic::{AtomicU64, Ordering};
@@ -32,7 +29,6 @@ pub struct IndexWriter {
file_path: String,
file: File,
index_size_bytes: Rc<AtomicU64>,
- fsync: bool,
}
// Safety: We are guaranteeing that IndexWriter will never be used from multiple threads
@@ -43,7 +39,6 @@ impl IndexWriter {
pub async fn new(
file_path: &str,
index_size_bytes: Rc<AtomicU64>,
- fsync: bool,
file_exists: bool,
) -> Result<Self, IggyError> {
let mut opts = OpenOptions::new();
@@ -82,48 +77,9 @@ impl IndexWriter {
file_path: file_path.to_string(),
file,
index_size_bytes,
- fsync,
})
}
- /// Appends multiple index buffer to the index file in a single operation.
- pub async fn save_indexes(&self, indexes: PooledBuffer) -> Result<(), IggyError> {
- if indexes.is_empty() {
- return Ok(());
- }
-
- let count = indexes.len() / INDEX_SIZE;
- let len = indexes.len();
-
- let position = self.index_size_bytes.load(Ordering::Relaxed);
- let file = &self.file;
- (&*file)
- .write_all_at(indexes, position)
- .await
- .0
- .error(|e: &std::io::Error| {
- format!(
- "Failed to write {} indexes to file: {}. {e}",
- count, self.file_path
- )
- })
- .map_err(|_| IggyError::CannotSaveIndexToSegment)?;
-
- self.index_size_bytes
- .fetch_add(len as u64, Ordering::Release);
-
- if self.fsync {
- let _ = self.fsync().await;
- }
- trace!(
- "Saved {count} indexes of size {} to file: {}",
- INDEX_SIZE * count,
- self.file_path
- );
-
- Ok(())
- }
-
pub fn size_counter(&self) -> Rc<AtomicU64> {
self.index_size_bytes.clone()
}
diff --git a/core/server_common/src/segment_storage/messages_reader.rs b/core/server_common/src/segment_storage/messages_reader.rs
index ecfe7e255..6c07e4533 100644
--- a/core/server_common/src/segment_storage/messages_reader.rs
+++ b/core/server_common/src/segment_storage/messages_reader.rs
@@ -15,136 +15,39 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{IggyIndexesMut, IggyMessagesBatchMut, PooledBuffer};
-use compio::buf::{IntoInner, IoBuf};
-use compio::fs::{File, OpenOptions};
-use compio::io::AsyncReadAtExt;
+use compio::fs::OpenOptions;
use err_trail::ErrContext;
use iggy_common::IggyError;
-use std::rc::Rc;
-use std::{
- io::ErrorKind,
- sync::atomic::{AtomicU64, Ordering},
-};
-use tracing::{error, trace};
+use tracing::trace;
-/// A dedicated struct for reading from the messages file.
+/// Path handle for a segment's messages file, validated openable at segment
+/// build. Reads go through the partition's own sealed-segment handles; this
+/// exists so storage plumbing (bootstrap, state transfer) can resolve the
+/// segment path.
#[derive(Debug)]
pub struct MessagesReader {
file_path: String,
- file: File,
- messages_size_bytes: Rc<AtomicU64>,
}
-// Safety: We are guaranteeing that MessagesReader will never be used from multiple threads
-unsafe impl Send for MessagesReader {}
-
impl MessagesReader {
- /// Opens the messages file in read mode.
- pub async fn new(
- file_path: &str,
- messages_size_bytes: Rc<AtomicU64>,
- ) -> Result<Self, IggyError> {
- let file = OpenOptions::new()
+ /// Opens the messages file read-only to prove it exists, then drops the
+ /// descriptor: nothing reads through this type.
+ pub async fn new(file_path: &str) -> Result<Self, IggyError> {
+ OpenOptions::new()
.read(true)
.open(file_path)
.await
.error(|e: &std::io::Error| format!("Failed to open messages file: {file_path}. {e}"))
.map_err(|_| IggyError::CannotReadFile)?;
- // posix_fadvise() is Linux-only in the nix crate
- #[cfg(target_os = "linux")]
- {
- let _ = nix::fcntl::posix_fadvise(
- &file,
- 0,
- 0, // 0 means the entire file
- nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL,
- )
- .info(|e: &nix::errno::Errno| {
- format!(
- "Failed to set sequential access pattern on messages file: {file_path}. {e}"
- )
- });
- }
-
- let size_bytes = messages_size_bytes.load(Ordering::Relaxed);
- trace!(
- "Opened messages file for reading: {file_path}, size: {}",
- size_bytes
- );
+ trace!("Validated messages file for reading: {file_path}");
Ok(Self {
file_path: file_path.to_string(),
- file,
- messages_size_bytes,
})
}
pub fn path(&self) -> String {
self.file_path.clone()
}
-
- /// Loads and returns a batch of messages from the messages file.
- pub async fn load_messages_from_disk(
- &self,
- indexes: IggyIndexesMut,
- ) -> Result<IggyMessagesBatchMut, IggyError> {
- let file_size = self.file_size();
- if file_size == 0 {
- return Ok(IggyMessagesBatchMut::empty());
- }
-
- let start_pos = indexes.base_position();
- let count_bytes = indexes.messages_size();
- let messages_count = indexes.count();
-
- if start_pos + count_bytes > file_size {
- return Ok(IggyMessagesBatchMut::empty());
- }
-
- let messages_bytes = match self.read_at(start_pos, count_bytes, true).await {
- Ok(buf) => buf,
- Err(e) if e.kind() == ErrorKind::UnexpectedEof => {
- return Ok(IggyMessagesBatchMut::empty());
- }
- Err(e) => {
- error!(
- "Error reading {messages_count} messages at position {start_pos} in file {} of size {}: {e}",
- self.file_path, file_size
- );
- return Err(IggyError::CannotReadMessage);
- }
- };
-
- Ok(IggyMessagesBatchMut::from_indexes_and_messages(
- indexes,
- messages_bytes,
- ))
- }
-
- /// Returns the size of the messages file in bytes.
- pub fn file_size(&self) -> u32 {
- self.messages_size_bytes.load(Ordering::Acquire) as u32
- }
-
- /// Reads `len` bytes from the messages file at the specified `offset`.
- async fn read_at(
- &self,
- offset: u32,
- len: u32,
- _use_pool: bool,
- ) -> Result<PooledBuffer, std::io::Error> {
- let buf = PooledBuffer::with_capacity(len as usize);
-
- let (result, buf) = self
- .file
- .read_exact_at(buf.slice(..len as usize), offset as u64)
- .await
- .into();
-
- let buf = buf.into_inner();
- result?;
- Ok(buf)
- }
}
diff --git a/core/server_common/src/segment_storage/messages_writer.rs b/core/server_common/src/segment_storage/messages_writer.rs
index 214c719b0..1d0094bec 100644
--- a/core/server_common/src/segment_storage/messages_writer.rs
+++ b/core/server_common/src/segment_storage/messages_writer.rs
@@ -15,10 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use super::write_batch_frozen;
use compio::fs::{File, OpenOptions};
use err_trail::ErrContext;
-use iggy_common::{IggyByteSize, IggyError, IggyMessagesBatch};
+use iggy_common::IggyError;
use std::{
rc::Rc,
sync::atomic::{AtomicU64, Ordering},
@@ -31,7 +30,6 @@ pub struct MessagesWriter {
file_path: String,
file: File,
messages_size_bytes: Rc<AtomicU64>,
- fsync: bool,
}
// Safety: We are guaranteeing that MessagesWriter will never be used from multiple threads
@@ -46,7 +44,6 @@ impl MessagesWriter {
pub async fn new(
file_path: &str,
messages_size_bytes: Rc<AtomicU64>,
- fsync: bool,
file_exists: bool,
) -> Result<Self, IggyError> {
let mut opts = OpenOptions::new();
@@ -90,38 +87,9 @@ impl MessagesWriter {
file_path: file_path.to_string(),
file,
messages_size_bytes,
- fsync,
})
}
- /// Append frozen (immutable) batches to the messages file.
- /// The caller retains the batches (for use in in-flight buffer) while disk I/O proceeds.
- pub async fn save_frozen_batches(
- &self,
- batches: &[IggyMessagesBatch],
- ) -> Result<IggyByteSize, IggyError> {
- let messages_size: u64 = batches.iter().map(|b| b.size() as u64).sum();
-
- let position = self.messages_size_bytes.load(Ordering::Relaxed);
- let file = &self.file;
- write_batch_frozen(file, position, batches)
- .await
- .error(|e: &IggyError| {
- format!(
- "Failed to write frozen batch to messages file: {}. {e}",
- self.file_path
- )
- })?;
-
- if self.fsync {
- let _ = self.fsync().await;
- }
-
- self.messages_size_bytes
- .fetch_add(messages_size, Ordering::Release);
-
- Ok(IggyByteSize::from(messages_size))
- }
pub fn path(&self) -> String {
self.file_path.clone()
}
diff --git a/core/server_common/src/segment_storage/mod.rs b/core/server_common/src/segment_storage/mod.rs
index 953b35908..e88d90fed 100644
--- a/core/server_common/src/segment_storage/mod.rs
+++ b/core/server_common/src/segment_storage/mod.rs
@@ -20,64 +20,14 @@ mod index_writer;
mod messages_reader;
mod messages_writer;
-use bytes::Bytes;
-use compio::{fs::File, io::AsyncWriteAtExt};
-use iggy_common::{IggyError, IggyMessagesBatch};
+use iggy_common::IggyError;
use std::rc::Rc;
-use tracing::error;
pub use index_reader::IndexReader;
pub use index_writer::IndexWriter;
pub use messages_reader::MessagesReader;
pub use messages_writer::MessagesWriter;
-/// Maximum number of IO vectors for a single writev() call.
-/// Linux typically has IOV_MAX=1024, but we use a conservative value to ensure
-/// cross-platform compatibility and leave room for any internal overhead.
-const MAX_IOV_COUNT: usize = 1024;
-
-/// Vectored write frozen (immutable) batches to file.
-pub async fn write_batch_frozen(
- file: &File,
- position: u64,
- batches: &[IggyMessagesBatch],
-) -> Result<usize, IggyError> {
- let (total_written, buffers) = batches.iter().fold(
- (0usize, Vec::with_capacity(batches.len())),
- |(size, mut bufs), batch| {
- bufs.push(batch.messages_bytes());
- (size + batch.size() as usize, bufs)
- },
- );
-
- write_vectored_chunked_bytes(file, position, buffers).await?;
- Ok(total_written)
-}
-
-/// Writes Bytes buffers to file using vectored I/O, chunking to respect IOV_MAX limits.
-async fn write_vectored_chunked_bytes(
- file: &File,
- mut position: u64,
- buffers: Vec<Bytes>,
-) -> Result<(), IggyError> {
- for chunk in buffers.chunks(MAX_IOV_COUNT) {
- let chunk_size: usize = chunk.iter().map(|b| b.len()).sum();
- let chunk_vec: Vec<Bytes> = chunk.to_vec();
-
- let (result, _) = (&*file)
- .write_vectored_all_at(chunk_vec, position)
- .await
- .into();
- result.map_err(|e| {
- error!("Failed to write frozen batch to messages file: {e}");
- IggyError::CannotWriteToFile
- })?;
-
- position += chunk_size as u64;
- }
- Ok(())
-}
-
unsafe impl Send for SegmentStorage {}
#[derive(Debug, Clone, Default)]
@@ -94,27 +44,21 @@ impl SegmentStorage {
index_path: &str,
messages_size: u64,
indexes_size: u64,
- log_fsync: bool,
- index_fsync: bool,
file_exists: bool,
) -> Result<Self, IggyError> {
let size = Rc::new(std::sync::atomic::AtomicU64::new(messages_size));
let indexes_size = Rc::new(std::sync::atomic::AtomicU64::new(indexes_size));
- let messages_writer = Rc::new(
- MessagesWriter::new(messages_path, size.clone(), log_fsync, file_exists).await?,
- );
+ let messages_writer = Rc::new(MessagesWriter::new(messages_path, size, file_exists).await?);
- let index_writer = Rc::new(
- IndexWriter::new(index_path, indexes_size.clone(), index_fsync, file_exists).await?,
- );
+ let index_writer = Rc::new(IndexWriter::new(index_path, indexes_size, file_exists).await?);
if file_exists {
messages_writer.fsync().await?;
index_writer.fsync().await?;
}
- let messages_reader = Rc::new(MessagesReader::new(messages_path, size).await?);
- let index_reader = Rc::new(IndexReader::new(index_path, indexes_size).await?);
+ let messages_reader = Rc::new(MessagesReader::new(messages_path).await?);
+ let index_reader = Rc::new(IndexReader::new(index_path).await?);
Ok(Self {
messages_writer: Some(messages_writer),
messages_reader: Some(messages_reader),
diff --git a/core/server_common/src/send_messages.rs b/core/server_common/src/send_messages.rs
index e7cfc9bab..ac40e544e 100644
--- a/core/server_common/src/send_messages.rs
+++ b/core/server_common/src/send_messages.rs
@@ -15,115 +15,88 @@
// specific language governing permissions and limitations
// under the License.
+//! Server-side handling of the canonical message batch (layout in
+//! [`iggy_binary_protocol::batch`]).
+//!
+//! Producers put the batch on the wire themselves; admission
+//! ([`convert_request_message`]) verifies the producer's checksums, strips the
+//! routing metadata, and stamps the partition. From there the same bytes are
+//! journaled, replicated, persisted, and served back to polls.
+
use crate::consensus_message::{MESSAGE_ALIGN, Message};
use crate::iobuf::Owned;
use crate::sharding::IggyNamespace;
use bytes::{Bytes, BytesMut};
-use iggy_binary_protocol::{PrepareHeader, RoutedRequestHeader};
-use iggy_common::{EncryptorKind, INDEX_SIZE, IggyError, random_id};
-use std::hash::Hasher;
+use iggy_binary_protocol::batch;
+use iggy_binary_protocol::requests::messages::SendMessagesHeader as SendMessagesMetadata;
+use iggy_binary_protocol::{PrepareHeader, RoutedRequestHeader, WireDecode, WireError};
+use iggy_common::{EncryptorKind, IggyError, random_id};
use twox_hash::XxHash3_64;
-pub const COMMAND_HEADER_SIZE: usize = 256;
-pub const PREPARE_SPLIT_POINT: usize = 512;
-const MESSAGE_HEADER_SIZE: usize = 48;
-const LEGACY_MESSAGE_HEADER_SIZE: usize = 64;
-const BATCH_CHECKSUM_OFFSET: usize = 40;
-const MESSAGE_COUNT_OFFSET: usize = 48;
-const MAX_TIMESTAMP_DELTA_MICROS: u64 = u32::MAX as u64;
-
-#[derive(Debug, Clone, Copy, Default)]
-pub struct SendMessagesHeader {
- pub partition_id: u64,
- pub base_offset: u64,
- pub base_timestamp: u64,
- pub origin_timestamp: u64,
- pub batch_length: u64,
- pub batch_checksum: u64,
- pub message_count: u32,
-}
-
-impl SendMessagesHeader {
- pub const fn new(
- partition_id: u64,
- origin_timestamp: u64,
- batch_length: u64,
- message_count: u32,
- ) -> Self {
- Self {
- partition_id,
- base_offset: 0,
- base_timestamp: 0,
- origin_timestamp,
- batch_length,
- batch_checksum: 0,
- message_count,
- }
- }
-
- pub fn decode(bytes: &[u8]) -> Result<Self, IggyError> {
- if bytes.len() < COMMAND_HEADER_SIZE {
- return Err(IggyError::InvalidCommand);
- }
-
- let batch_length = read_u64(bytes, 32)?;
- if batch_length < COMMAND_HEADER_SIZE as u64 {
- return Err(IggyError::InvalidCommand);
- }
-
- Ok(Self {
- partition_id: read_u64(bytes, 0)?,
- base_offset: read_u64(bytes, 8)?,
- base_timestamp: read_u64(bytes, 16)?,
- origin_timestamp: read_u64(bytes, 24)?,
- batch_length,
- batch_checksum: read_u64(bytes, BATCH_CHECKSUM_OFFSET)?,
- message_count: read_u32(bytes, MESSAGE_COUNT_OFFSET)?,
- })
- }
+pub use iggy_binary_protocol::batch::{
+ BATCH_CHECKSUM_OFFSET, BATCH_HEADER_SIZE, BATCH_MESSAGE_HEADER_SIZE, BatchHeader,
+ BatchIntegrity, BatchIterator, BatchIteratorWithOffsets, BatchMessageHeader, BatchMessageView,
+ BatchMessageViewWithOffsets, BatchRef, MAX_TIMESTAMP_DELTA_MICROS, calculate_batch_checksum,
+};
- pub fn encode_into(&self, bytes: &mut [u8]) {
- assert!(bytes.len() >= COMMAND_HEADER_SIZE);
- bytes[..COMMAND_HEADER_SIZE].fill(0);
- bytes[0..8].copy_from_slice(&self.partition_id.to_le_bytes());
- bytes[8..16].copy_from_slice(&self.base_offset.to_le_bytes());
- bytes[16..24].copy_from_slice(&self.base_timestamp.to_le_bytes());
- bytes[24..32].copy_from_slice(&self.origin_timestamp.to_le_bytes());
- bytes[32..40].copy_from_slice(&self.batch_length.to_le_bytes());
- bytes[BATCH_CHECKSUM_OFFSET..BATCH_CHECKSUM_OFFSET + 8]
- .copy_from_slice(&self.batch_checksum.to_le_bytes());
- bytes[MESSAGE_COUNT_OFFSET..MESSAGE_COUNT_OFFSET + 4]
- .copy_from_slice(&self.message_count.to_le_bytes());
- }
+/// Size of the batch header at the front of a `SendMessages` prepare body.
+pub const COMMAND_HEADER_SIZE: usize = BATCH_HEADER_SIZE;
- pub fn total_size(&self) -> usize {
- usize::try_from(self.batch_length).expect("batch length exceeds usize::MAX")
- }
+/// Offset of the blob inside a prepare frame:
+/// `[256B PrepareHeader][256B batch header][blob]`.
+pub const PREPARE_SPLIT_POINT: usize = 512;
- pub fn blob_len(&self) -> Result<usize, IggyError> {
- usize::try_from(
- self.batch_length
- .checked_sub(COMMAND_HEADER_SIZE as u64)
- .ok_or(IggyError::InvalidCommand)?,
- )
- .map_err(|_| IggyError::InvalidCommand)
+/// Map a wire-level batch error onto the typed `IggyError` the server paths
+/// key on. The two integrity variants keep their payloads; every structural
+/// error is an invalid command.
+fn batch_error(error: &WireError) -> IggyError {
+ match *error {
+ WireError::InvalidBatchChecksum {
+ stored,
+ computed,
+ base_offset,
+ } => IggyError::InvalidBatchChecksum(stored, computed, base_offset),
+ WireError::InvalidMessageChecksum {
+ stored,
+ computed,
+ offset,
+ } => IggyError::InvalidMessageChecksum(stored, computed, offset),
+ _ => IggyError::InvalidCommand,
}
+}
- pub fn into_frozen(self) -> FrozenBatchHeader {
- let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(COMMAND_HEADER_SIZE);
- self.encode_into(buffer.as_mut_slice());
- buffer.into()
- }
+/// Decode one batch record, verifying the batch checksum and every
+/// per-message checksum.
+///
+/// See [`batch::decode_batch_slice`]; this wrapper types the error for the
+/// server paths.
+///
+/// # Errors
+/// [`IggyError::InvalidCommand`] for a short or inconsistent record, and
+/// [`IggyError::InvalidBatchChecksum`] / [`IggyError::InvalidMessageChecksum`]
+/// on an integrity mismatch.
+pub fn decode_batch_slice(body: &[u8]) -> Result<BatchRef<'_>, IggyError> {
+ batch::decode_batch_slice(body).map_err(|error| batch_error(&error))
+}
- #[must_use]
- pub fn checksum_for_blob(&self, blob: &[u8]) -> u64 {
- calculate_batch_checksum(self, blob)
- }
+/// [`decode_batch_slice`] with the integrity level chosen by the caller.
+///
+/// The one caller that passes anything but [`BatchIntegrity::Verify`] is the
+/// disk poll under its operator knob; layout checks are not optional either
+/// way.
+///
+/// # Errors
+/// See [`decode_batch_slice`].
+pub fn decode_batch_slice_with(
+ body: &[u8],
+ integrity: BatchIntegrity,
+) -> Result<BatchRef<'_>, IggyError> {
+ batch::decode_batch_slice_with(body, integrity).map_err(|error| batch_error(&error))
}
#[derive(Debug, Clone)]
pub struct SendMessagesOwned {
- pub header: SendMessagesHeader,
+ pub header: BatchHeader,
pub blob: Bytes,
}
@@ -166,7 +139,7 @@ impl SendMessagesOwned {
let payload_length =
u32::try_from(message.payload.len()).map_err(|_| IggyError::InvalidCommand)?;
- let mut header = [0u8; MESSAGE_HEADER_SIZE];
+ let mut header = [0u8; BATCH_MESSAGE_HEADER_SIZE];
header[8..24].copy_from_slice(&id.to_le_bytes());
header[24..28].copy_from_slice(&offset_delta.to_le_bytes());
header[28..32].copy_from_slice(×tamp_delta.to_le_bytes());
@@ -182,7 +155,7 @@ impl SendMessagesOwned {
}
let blob = blob.freeze();
- let mut header = SendMessagesHeader::new(
+ let mut header = BatchHeader::new(
namespace.partition_id() as u64,
origin_timestamp,
u64::try_from(COMMAND_HEADER_SIZE + blob.len())
@@ -199,9 +172,9 @@ impl SendMessagesOwned {
mut request_header: RoutedRequestHeader,
) -> Result<Message<RoutedRequestHeader>, IggyError> {
let total_size = std::mem::size_of::<RoutedRequestHeader>() + self.header.total_size();
- // The converted body differs in size from the legacy wire body the
- // header described; a stale `size` truncates the rebuilt blob for
- // every downstream slice (stamping, journal reads).
+ // The rebuilt body differs in size from the wire body the header
+ // described; a stale `size` truncates the blob for every downstream
+ // slice (stamping, journal reads).
request_header.size = u32::try_from(total_size).map_err(|_| IggyError::InvalidCommand)?;
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total_size);
let bytes = buffer.as_mut_slice();
@@ -308,163 +281,21 @@ impl<'a> IntoIterator for &'a IggyMessages {
}
}
-#[derive(Debug, Clone, Copy)]
-pub struct SendMessagesRef<'a> {
- pub header: SendMessagesHeader,
- blob: &'a [u8],
-}
-
-#[allow(dead_code)]
-impl<'a> SendMessagesRef<'a> {
- pub const fn iter(&self) -> SendMessagesIterator<'a> {
- SendMessagesIterator {
- blob: self.blob,
- position: 0,
- }
- }
-
- pub const fn iter_with_offsets(&self) -> SendMessagesIteratorWithOffsets<'a> {
- SendMessagesIteratorWithOffsets {
- blob: self.blob,
- position: 0,
- }
- }
-
- pub const fn blob(&self) -> &'a [u8] {
- self.blob
- }
-
- pub const fn message_count(&self) -> u32 {
- self.header.message_count
- }
-}
-
-#[allow(dead_code)]
-#[derive(Debug, Clone, Copy)]
-pub struct SendMessagesMessageHeader {
- pub checksum: u64,
- pub id: u128,
- pub offset_delta: u32,
- /// Microsecond delta from `SendMessagesHeader::origin_timestamp`.
- ///
- /// This is stored in `u32`, which limits a single batch to roughly
- /// 71.6 minutes of origin timestamp span.
- pub timestamp_delta: u32,
- pub user_headers_length: u32,
- pub payload_length: u32,
-}
-
-impl SendMessagesMessageHeader {
- fn decode(bytes: &[u8]) -> Result<Self, IggyError> {
- if bytes.len() < MESSAGE_HEADER_SIZE {
- return Err(IggyError::InvalidCommand);
- }
-
- let reserved = read_u64(bytes, 40)?;
- if reserved != 0 {
- return Err(IggyError::InvalidCommand);
- }
-
- Ok(Self {
- checksum: read_u64(bytes, 0)?,
- id: read_u128(bytes, 8)?,
- offset_delta: read_u32(bytes, 24)?,
- timestamp_delta: read_u32(bytes, 28)?,
- user_headers_length: read_u32(bytes, 32)?,
- payload_length: read_u32(bytes, 36)?,
- })
- }
-
- const fn total_size(&self) -> usize {
- MESSAGE_HEADER_SIZE + self.user_headers_length as usize + self.payload_length as usize
- }
-}
-
-#[allow(dead_code)]
-#[derive(Debug, Clone, Copy)]
-pub struct SendMessagesMessageView<'a> {
- pub header: SendMessagesMessageHeader,
- pub user_headers: &'a [u8],
- pub payload: &'a [u8],
-}
-
-#[allow(dead_code)]
-pub struct SendMessagesIterator<'a> {
- blob: &'a [u8],
- position: usize,
-}
-
-impl<'a> Iterator for SendMessagesIterator<'a> {
- type Item = SendMessagesMessageView<'a>;
-
- fn next(&mut self) -> Option<Self::Item> {
- if self.position >= self.blob.len() {
- return None;
- }
-
- let header = SendMessagesMessageHeader::decode(&self.blob[self.position..]).ok()?;
- let start = self.position + MESSAGE_HEADER_SIZE;
- let payload_end = start + header.payload_length as usize;
- let headers_end = payload_end + header.user_headers_length as usize;
- let payload = self.blob.get(start..payload_end)?;
- let user_headers = self.blob.get(payload_end..headers_end)?;
- self.position += header.total_size();
- Some(SendMessagesMessageView {
- header,
- user_headers,
- payload,
- })
- }
-}
-
-#[derive(Debug, Clone, Copy)]
-pub struct SendMessagesMessageViewWithOffsets<'a> {
- pub message: SendMessagesMessageView<'a>,
- pub start: usize,
- pub end: usize,
-}
-
-pub struct SendMessagesIteratorWithOffsets<'a> {
- blob: &'a [u8],
- position: usize,
+/// Encode `header` into its own frozen 256-byte buffer (e.g. the rewritten
+/// header fragment of a server-sliced poll batch).
+#[must_use]
+pub fn frozen_batch_header(header: &BatchHeader) -> crate::iobuf::Frozen<MESSAGE_ALIGN> {
+ let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(COMMAND_HEADER_SIZE);
+ header.encode_into(buffer.as_mut_slice());
+ buffer.into()
}
-impl<'a> Iterator for SendMessagesIteratorWithOffsets<'a> {
- type Item = SendMessagesMessageViewWithOffsets<'a>;
-
- fn next(&mut self) -> Option<Self::Item> {
- if self.position >= self.blob.len() {
- return None;
- }
-
- let start = self.position;
- let header = SendMessagesMessageHeader::decode(&self.blob[self.position..]).ok()?;
- let message_start = self.position + MESSAGE_HEADER_SIZE;
- let payload_end = message_start + header.payload_length as usize;
- let headers_end = payload_end + header.user_headers_length as usize;
- let payload = self.blob.get(message_start..payload_end)?;
- let user_headers = self.blob.get(payload_end..headers_end)?;
- self.position += header.total_size();
- Some(SendMessagesMessageViewWithOffsets {
- message: SendMessagesMessageView {
- header,
- user_headers,
- payload,
- },
- start,
- end: self.position,
- })
- }
-}
-
-pub(crate) type FrozenBatchHeader = crate::iobuf::Frozen<MESSAGE_ALIGN>;
-
/// Re-encode a canonical `SendMessages` request with every message's payload
/// and user headers encrypted, per-message checksums and lengths recomputed,
/// and the batch header (length + checksum) restamped.
///
/// Runs ONCE, on the primary at ingestion (after [`convert_request_message`]
-/// canonicalized the wire form), so the ciphertext is what replicates: every
+/// admitted the wire form), so the ciphertext is what replicates: every
/// replica journals and persists identical bytes, and the poll path decrypts
/// uniformly regardless of which replica or tier served the fragment.
///
@@ -495,7 +326,7 @@ pub fn encrypt_batch_request(
let user_headers_length =
u32::try_from(user_headers.len()).map_err(|_| IggyError::InvalidCommand)?;
- let mut header = [0u8; MESSAGE_HEADER_SIZE];
+ let mut header = [0u8; BATCH_MESSAGE_HEADER_SIZE];
header[8..24].copy_from_slice(&view.header.id.to_le_bytes());
header[24..28].copy_from_slice(&view.header.offset_delta.to_le_bytes());
header[28..32].copy_from_slice(&view.header.timestamp_delta.to_le_bytes());
@@ -518,23 +349,111 @@ pub fn encrypt_batch_request(
SendMessagesOwned { header, blob }.encode_request(request_header)
}
-/// Whether the legacy transcode stamps a batch checksum onto its output.
+/// Rebuild one stored batch record with every message's payload and user
+/// headers decrypted: the poll reply's single decrypt point, mirroring
+/// [`encrypt_batch_request`]. Lengths, per-message checksums, and the batch
+/// header (length + checksum) are restamped over the plaintext.
+///
+/// Framing is validated layout-only: the read path already applied its
+/// integrity knob to the stored bytes, and a server-sliced fragment's header
+/// was rewritten checksum-consistent at slice time.
+///
+/// # Errors
+/// [`IggyError::InvalidCommand`] on a malformed record;
+/// [`IggyError::CannotDecryptData`] when a section fails to decrypt.
+pub fn decrypt_batch_record(
+ record: &[u8],
+ encryptor: &EncryptorKind,
+) -> Result<Vec<u8>, IggyError> {
+ let batch = decode_batch_slice_with(record, BatchIntegrity::LayoutOnly)?;
+ if record.len() != batch.header.total_size() {
+ return Err(IggyError::InvalidCommand);
+ }
+
+ let mut blob = BytesMut::with_capacity(batch.blob().len());
+ for view in batch.iter() {
+ let payload = encryptor
+ .decrypt(view.payload)
+ .map_err(|_| IggyError::CannotDecryptData)?;
+ let decrypted_user_headers = if view.user_headers.is_empty() {
+ None
+ } else {
+ Some(
+ encryptor
+ .decrypt(view.user_headers)
+ .map_err(|_| IggyError::CannotDecryptData)?,
+ )
+ };
+ let user_headers: &[u8] = decrypted_user_headers.as_deref().unwrap_or_default();
+ let payload_length = u32::try_from(payload.len()).map_err(|_| IggyError::InvalidCommand)?;
+ let user_headers_length =
+ u32::try_from(user_headers.len()).map_err(|_| IggyError::InvalidCommand)?;
+
+ let mut header = [0u8; BATCH_MESSAGE_HEADER_SIZE];
+ header[8..24].copy_from_slice(&view.header.id.to_le_bytes());
+ header[24..28].copy_from_slice(&view.header.offset_delta.to_le_bytes());
+ header[28..32].copy_from_slice(&view.header.timestamp_delta.to_le_bytes());
+ header[32..36].copy_from_slice(&user_headers_length.to_le_bytes());
+ header[36..40].copy_from_slice(&payload_length.to_le_bytes());
+ let msg_start = blob.len();
+ blob.extend_from_slice(&header);
+ blob.extend_from_slice(&payload);
+ blob.extend_from_slice(user_headers);
+ let checksum = XxHash3_64::oneshot(&blob[msg_start + 8..]);
+ blob[msg_start..msg_start + 8].copy_from_slice(&checksum.to_le_bytes());
+ }
+
+ let mut header = batch.header;
+ header.batch_length =
+ u64::try_from(COMMAND_HEADER_SIZE + blob.len()).map_err(|_| IggyError::InvalidCommand)?;
+ header.batch_checksum = calculate_batch_checksum(&header, &blob);
+
+ let mut out = vec![0u8; COMMAND_HEADER_SIZE + blob.len()];
+ header.encode_into(&mut out[..COMMAND_HEADER_SIZE]);
+ out[COMMAND_HEADER_SIZE..].copy_from_slice(&blob);
+ Ok(out)
+}
+
+/// Whether admission stamps a batch checksum onto its output.
///
/// The recompute is an `XxHash3` batch-checksum pass, needed only when a reader
-/// validates the transcoded batch before [`stamp_prepare_for_persistence`]
-/// recomputes it: the encrypt ingest path re-decodes the canonicalized batch
+/// validates the admitted batch before [`stamp_prepare_for_persistence`]
+/// recomputes it: the encrypt ingest path re-decodes the admitted batch
/// (`encrypt_batch_request`'s validating decode, then the second `convert` its
-/// output re-enters as the canonical-vs-legacy discriminator). The partition
-/// ingest path has no such reader, so it skips the pass and the checksum stays
-/// zero until stamp.
+/// output re-enters as the canonical-batch fast path). The partition ingest
+/// path has no such reader, so it skips the pass and the checksum stays zero
+/// until stamp.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChecksumMode {
- /// Compute the batch checksum for the transcoded batch.
+ /// Compute the batch checksum for the admitted batch.
Compute,
/// Leave the batch checksum zero; `stamp_prepare_for_persistence` fills it.
Skip,
}
+/// Admit a `SendMessages` request into the pipeline form
+/// `[RoutedRequestHeader][256B batch header][blob]`.
+///
+/// Two input shapes reach this:
+/// - The wire form `[metadata][batch]` from a producer. The metadata section
+/// is validated and stripped, the batch is checksum-verified exactly as the
+/// producer hashed it (`partition_id` still zero), and the partition is
+/// stamped afterwards.
+/// - A body that already IS one canonical batch: the encrypt ingest path
+/// re-enters here with its own output. Detected first via the
+/// checksum-verified decode - a wire-form body cannot pass it, since its
+/// leading metadata bytes cannot form a batch whose length and checksum
+/// both match.
+///
+/// Either way the batch must fill the body exactly: `size` and `batch_length`
+/// are independent client-supplied fields, and a suffix past `batch_length`
+/// is covered by no checksum, still rides the buffer to disk, and desyncs the
+/// segment walk that advances by `batch_length`.
+///
+/// # Errors
+/// [`IggyError::InvalidCommand`] on a malformed body, an empty batch, or a
+/// metadata/batch count mismatch; typed checksum errors from the validating
+/// decode.
pub fn convert_request_message(
namespace: IggyNamespace,
message: Message<RoutedRequestHeader>,
@@ -543,205 +462,78 @@ pub fn convert_request_message(
let request_header = *message.header();
let total_size = request_header.size as usize;
let body = &message.as_slice()[std::mem::size_of::<RoutedRequestHeader>()..total_size];
- // A canonical body enters the pipeline verbatim, so it must end exactly at
- // `batch_length`: `size` and `batch_length` are independent client-supplied
- // fields and `decode_batch_slice` only lower-bounds the frame. A suffix past
- // `batch_length` is covered by no checksum, still rides the buffer to disk,
- // and desyncs the segment walk that advances by `batch_length`.
- match decode_batch_slice(body) {
- Ok(batch) if batch.message_count() == 0 => Err(IggyError::InvalidCommand),
- Ok(batch) if body.len() == batch.header.total_size() => Ok(message),
- Ok(_) => Err(IggyError::InvalidCommand),
- Err(_) => transcode_legacy_request(namespace, body, request_header, checksum),
+
+ if let Ok(batch) = decode_batch_slice(body) {
+ if batch.message_count() == 0 || body.len() != batch.header.total_size() {
+ return Err(IggyError::InvalidCommand);
+ }
+ return Ok(message);
}
+
+ admit_wire_request(namespace, body, request_header, checksum)
}
-/// Transcode a legacy `SendMessages` request body directly into the canonical
-/// `[RoutedRequestHeader][256B SendMessagesHeader][blob]` form, writing each message
-/// record straight into the final aligned buffer.
-///
-/// Fused replacement for the `from_legacy_request(..).encode_request(..)`
-/// two-step: a size walk over the legacy input sizes the single output
-/// allocation, then a write walk lays down each canonical record in place. This
-/// drops the intermediate blob allocation and the full-blob copy the two-step
-/// paid. Output bytes are identical to that path.
-///
-/// `checksum` selects whether the output carries a batch checksum (see
-/// [`ChecksumMode`]); [`ChecksumMode::Skip`] leaves it zero for the partition
-/// ingest path, where stamp recomputes it.
-fn transcode_legacy_request(
+/// Validate a producer's `[metadata][batch]` body and rebuild it as the
+/// pipeline form with the partition stamped.
+fn admit_wire_request(
namespace: IggyNamespace,
body: &[u8],
mut request_header: RoutedRequestHeader,
checksum: ChecksumMode,
) -> Result<Message<RoutedRequestHeader>, IggyError> {
- let (message_count, messages) = legacy_messages_slice(body)?;
- if message_count == 0 {
+ if body.len() < 4 {
return Err(IggyError::InvalidCommand);
}
- let mut parsed = Vec::with_capacity(message_count as usize);
- let mut origin_timestamp = u64::MAX;
- let mut cursor = 0usize;
- let mut blob_len = 0usize;
-
- while cursor < messages.len() && parsed.len() < message_count as usize {
- let legacy = LegacyMessageRef::decode(&messages[cursor..])?;
- origin_timestamp = origin_timestamp.min(legacy.origin_timestamp);
- cursor += legacy.total_size;
- blob_len = blob_len
- .checked_add(MESSAGE_HEADER_SIZE + legacy.payload.len() + legacy.user_headers.len())
- .ok_or(IggyError::InvalidCommand)?;
- parsed.push(legacy);
+ let metadata_length = u32::from_le_bytes(
+ body[..4]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ) as usize;
+ let batch_start = 4usize
+ .checked_add(metadata_length)
+ .ok_or(IggyError::InvalidCommand)?;
+ if body.len() < batch_start {
+ return Err(IggyError::InvalidCommand);
}
- if parsed.len() != message_count as usize || cursor != messages.len() {
+ let (metadata, consumed) = SendMessagesMetadata::decode(&body[4..batch_start])
+ .map_err(|_| IggyError::InvalidCommand)?;
+ if consumed != metadata_length {
return Err(IggyError::InvalidCommand);
}
- if origin_timestamp == u64::MAX {
- origin_timestamp = 0;
+ let batch_bytes = &body[batch_start..];
+ let batch = decode_batch_slice(batch_bytes)?;
+ if batch.message_count() == 0
+ || batch.message_count() != metadata.messages_count
+ || batch_bytes.len() != batch.header.total_size()
+ {
+ return Err(IggyError::InvalidCommand);
}
let header_size = std::mem::size_of::<RoutedRequestHeader>();
- let batch_length = COMMAND_HEADER_SIZE
- .checked_add(blob_len)
- .ok_or(IggyError::InvalidCommand)?;
- let total_size = header_size
- .checked_add(batch_length)
- .ok_or(IggyError::InvalidCommand)?;
+ let total_size = header_size + batch.header.total_size();
request_header.size = u32::try_from(total_size).map_err(|_| IggyError::InvalidCommand)?;
-
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total_size);
let bytes = buffer.as_mut_slice();
bytes[0..header_size].copy_from_slice(bytemuck::bytes_of(&request_header));
-
- let mut write = PREPARE_SPLIT_POINT;
- for (index, legacy) in parsed.iter().enumerate() {
- let id = if legacy.id == 0 {
- random_id::get_uuid()
- } else {
- legacy.id
- };
- let offset_delta = u32::try_from(index).map_err(|_| IggyError::InvalidCommand)?;
- let timestamp_delta = legacy
- .origin_timestamp
- .checked_sub(origin_timestamp)
- .ok_or(IggyError::InvalidCommand)?;
- if timestamp_delta > MAX_TIMESTAMP_DELTA_MICROS {
- return Err(IggyError::InvalidMessageTimestampDelta(timestamp_delta));
+ bytes[header_size..total_size].copy_from_slice(batch_bytes);
+
+ // The producer hashed `partition_id = 0`; stamp the resolved partition
+ // and restamp (or clear, for the stamp-fills-it path) the batch checksum.
+ let mut stamped = batch.header;
+ stamped.partition_id = namespace.partition_id() as u64;
+ stamped.batch_checksum = match checksum {
+ ChecksumMode::Compute => {
+ calculate_batch_checksum(&stamped, &bytes[PREPARE_SPLIT_POINT..total_size])
}
- let timestamp_delta =
- u32::try_from(timestamp_delta).map_err(|_| IggyError::InvalidCommand)?;
- let user_headers_length =
- u32::try_from(legacy.user_headers.len()).map_err(|_| IggyError::InvalidCommand)?;
- let payload_length =
- u32::try_from(legacy.payload.len()).map_err(|_| IggyError::InvalidCommand)?;
-
- let mut header = [0u8; MESSAGE_HEADER_SIZE];
- header[8..24].copy_from_slice(&id.to_le_bytes());
- header[24..28].copy_from_slice(&offset_delta.to_le_bytes());
- header[28..32].copy_from_slice(×tamp_delta.to_le_bytes());
- header[32..36].copy_from_slice(&user_headers_length.to_le_bytes());
- header[36..40].copy_from_slice(&payload_length.to_le_bytes());
- let msg_start = write;
- bytes[write..write + MESSAGE_HEADER_SIZE].copy_from_slice(&header);
- write += MESSAGE_HEADER_SIZE;
- bytes[write..write + legacy.payload.len()].copy_from_slice(legacy.payload);
- write += legacy.payload.len();
- bytes[write..write + legacy.user_headers.len()].copy_from_slice(legacy.user_headers);
- write += legacy.user_headers.len();
- // The cover is [msg_start + 8 .. write], including the 8 reserved zero
- // header bytes. This relies on the stack header being zero-initialized,
- // not on the output buffer being pre-zeroed.
- let checksum = XxHash3_64::oneshot(&bytes[msg_start + 8..write]);
- bytes[msg_start..msg_start + 8].copy_from_slice(&checksum.to_le_bytes());
- }
-
- let mut command = SendMessagesHeader::new(
- namespace.partition_id() as u64,
- origin_timestamp,
- batch_length as u64,
- message_count,
- );
- if checksum == ChecksumMode::Compute {
- command.batch_checksum = calculate_batch_checksum(
- &command,
- &bytes[PREPARE_SPLIT_POINT..PREPARE_SPLIT_POINT + blob_len],
- );
- }
- command.encode_into(&mut bytes[header_size..header_size + COMMAND_HEADER_SIZE]);
+ ChecksumMode::Skip => 0,
+ };
+ stamped.encode_into(&mut bytes[header_size..header_size + COMMAND_HEADER_SIZE]);
Message::try_from(buffer).map_err(|_| IggyError::InvalidCommand)
}
-/// Decode one batch slice (`[256B command header][blob]`), validating the batch
-/// checksum and every per-message checksum. The persisted segment-file record
-/// and the request slice share this layout, so both decode through here.
-///
-/// `body` may extend past the batch: the poll disk walk hands in the rest of the
-/// chunk and steps by `batch_length`. Callers whose buffer is meant to BE the
-/// batch must reject the surplus themselves - see [`convert_request_message`].
-pub fn decode_batch_slice(body: &[u8]) -> Result<SendMessagesRef<'_>, IggyError> {
- decode_batch_slice_with(body, BatchIntegrity::Verify)
-}
-
-/// How much of a batch record [`decode_batch_slice_with`] proves before returning it.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum BatchIntegrity {
- /// Re-hash the batch and reject it unless it matches its own `batch_checksum`.
- Verify,
- /// Check the framing only, and hand back whatever it describes. The caller is
- /// accepting bytes that may not be the ones written.
- LayoutOnly,
-}
-
-/// [`decode_batch_slice`] with the integrity level chosen by the caller.
-///
-/// An enum, not a bool: the one caller that passes anything but
-/// [`BatchIntegrity::Verify`] is the disk poll under its operator knob, and
-/// `..., false)` there reads like a detail rather than opting a read out of
-/// corruption detection.
-///
-/// Layout checks are not optional either way: a short or self-inconsistent record is
-/// rejected regardless, because the caller would otherwise index past it.
-///
-/// # Errors
-/// [`IggyError::InvalidCommand`] for a short or inconsistent record, and
-/// [`IggyError::InvalidBatchChecksum`] under [`BatchIntegrity::Verify`] when the batch
-/// does not match. Callers that must tell corruption from a partial tail need both.
-pub fn decode_batch_slice_with(
- body: &[u8],
- integrity: BatchIntegrity,
-) -> Result<SendMessagesRef<'_>, IggyError> {
- if body.len() < COMMAND_HEADER_SIZE {
- return Err(IggyError::InvalidCommand);
- }
-
- let header = SendMessagesHeader::decode(&body[..COMMAND_HEADER_SIZE])?;
- let blob_len = header.blob_len()?;
- if body.len() < header.total_size() {
- return Err(IggyError::InvalidCommand);
- }
-
- let blob = &body[COMMAND_HEADER_SIZE..COMMAND_HEADER_SIZE + blob_len];
- let batch = SendMessagesRef { header, blob };
- match integrity {
- BatchIntegrity::Verify => {
- let expected_checksum = verify_and_recompute_batch_checksum(&batch)?;
- if header.batch_checksum != expected_checksum {
- return Err(IggyError::InvalidBatchChecksum(
- header.batch_checksum,
- expected_checksum,
- header.base_offset,
- ));
- }
- }
- BatchIntegrity::LayoutOnly => validate_batch_layout(&batch)?,
- }
-
- Ok(batch)
-}
-
/// Decode a `Prepare` message from a slice of bytes, validating the batch
/// checksum and every per-message checksum.
///
@@ -755,7 +547,7 @@ pub fn decode_batch_slice_with(
/// outside `[header_size, bytes.len()]`, a `size` that does not describe the
/// batch exactly, or frames that do not tile the batch;
/// `InvalidBatchChecksum` / `InvalidMessageChecksum` on an integrity mismatch.
-pub fn decode_prepare_slice(bytes: &[u8]) -> Result<SendMessagesRef<'_>, IggyError> {
+pub fn decode_prepare_slice(bytes: &[u8]) -> Result<BatchRef<'_>, IggyError> {
decode_prepare_slice_inner(bytes, true)
}
@@ -778,14 +570,14 @@ pub fn decode_prepare_slice(bytes: &[u8]) -> Result<SendMessagesRef<'_>, IggyErr
///
/// Same structural errors as [`decode_prepare_slice`], minus
/// `InvalidBatchChecksum` and `InvalidMessageChecksum`.
-pub fn decode_prepare_slice_trusted(bytes: &[u8]) -> Result<SendMessagesRef<'_>, IggyError> {
+pub fn decode_prepare_slice_trusted(bytes: &[u8]) -> Result<BatchRef<'_>, IggyError> {
decode_prepare_slice_inner(bytes, false)
}
fn decode_prepare_slice_inner(
bytes: &[u8],
validate_checksum: bool,
-) -> Result<SendMessagesRef<'_>, IggyError> {
+) -> Result<BatchRef<'_>, IggyError> {
let header_size = std::mem::size_of::<PrepareHeader>();
if bytes.len() < header_size {
return Err(IggyError::InvalidCommand);
@@ -814,8 +606,9 @@ fn decode_prepare_slice_inner(
return Err(IggyError::InvalidCommand);
}
- let header = SendMessagesHeader::decode(&body[..COMMAND_HEADER_SIZE])?;
- let blob_len = header.blob_len()?;
+ let header =
+ BatchHeader::decode(&body[..COMMAND_HEADER_SIZE]).map_err(|error| batch_error(&error))?;
+ let blob_len = header.blob_len().map_err(|error| batch_error(&error))?;
// Exact, not a lower bound: a prepare frame IS one batch, so bytes past
// `batch_length` belong to nobody - no checksum covers them, yet the flush
// writes them, desyncing the segment walk. Readers walking a multi-batch
@@ -825,9 +618,10 @@ fn decode_prepare_slice_inner(
}
let blob = &body[COMMAND_HEADER_SIZE..COMMAND_HEADER_SIZE + blob_len];
- let batch = SendMessagesRef { header, blob };
+ let batch = BatchRef::new(header, blob);
if validate_checksum {
- let expected_checksum = verify_and_recompute_batch_checksum(&batch)?;
+ let expected_checksum = batch::verify_and_recompute_batch_checksum(&batch)
+ .map_err(|error| batch_error(&error))?;
if header.batch_checksum != expected_checksum {
return Err(IggyError::InvalidBatchChecksum(
header.batch_checksum,
@@ -844,7 +638,7 @@ pub fn stamp_prepare_for_persistence(
mut message: Message<PrepareHeader>,
base_offset: u64,
base_timestamp: u64,
-) -> Result<(Message<PrepareHeader>, SendMessagesHeader, u32), IggyError> {
+) -> Result<(Message<PrepareHeader>, BatchHeader, u32), IggyError> {
let total_size = message.header().size as usize;
let bytes = message.as_mut_slice();
if bytes.len() < PREPARE_SPLIT_POINT || total_size < PREPARE_SPLIT_POINT {
@@ -853,7 +647,8 @@ pub fn stamp_prepare_for_persistence(
let header_offset = std::mem::size_of::<PrepareHeader>();
let mut command =
- SendMessagesHeader::decode(&bytes[header_offset..header_offset + COMMAND_HEADER_SIZE])?;
+ BatchHeader::decode(&bytes[header_offset..header_offset + COMMAND_HEADER_SIZE])
+ .map_err(|error| batch_error(&error))?;
command.base_offset = base_offset;
command.base_timestamp = base_timestamp;
let blob = &bytes[PREPARE_SPLIT_POINT..total_size];
@@ -862,200 +657,13 @@ pub fn stamp_prepare_for_persistence(
Ok((message, command, command.message_count))
}
-fn legacy_messages_slice(body: &[u8]) -> Result<(u32, &[u8]), IggyError> {
- if body.len() < 4 {
- return Err(IggyError::InvalidCommand);
- }
-
- let metadata_length = read_u32(body, 0)? as usize;
- let metadata_end = 4usize
- .checked_add(metadata_length)
- .ok_or(IggyError::InvalidCommand)?;
- if metadata_end < 4 || body.len() < metadata_end {
- return Err(IggyError::InvalidCommand);
- }
-
- let message_count = read_u32(body, metadata_end - 4)?;
- let indexes_len = usize::try_from(message_count)
- .ok()
- .and_then(|count| count.checked_mul(INDEX_SIZE))
- .ok_or(IggyError::InvalidCommand)?;
- let messages_start = metadata_end
- .checked_add(indexes_len)
- .ok_or(IggyError::InvalidCommand)?;
- if body.len() < messages_start {
- return Err(IggyError::InvalidCommand);
- }
-
- Ok((message_count, &body[messages_start..]))
-}
-
-#[derive(Clone, Copy)]
-struct LegacyMessageRef<'a> {
- id: u128,
- origin_timestamp: u64,
- user_headers: &'a [u8],
- payload: &'a [u8],
- total_size: usize,
-}
-
-impl<'a> LegacyMessageRef<'a> {
- fn decode(bytes: &'a [u8]) -> Result<Self, IggyError> {
- if bytes.len() < LEGACY_MESSAGE_HEADER_SIZE {
- return Err(IggyError::InvalidCommand);
- }
-
- let user_headers_length = read_u32(bytes, 48)? as usize;
- let payload_length = read_u32(bytes, 52)? as usize;
- let total_size = LEGACY_MESSAGE_HEADER_SIZE
- .checked_add(payload_length)
- .and_then(|size| size.checked_add(user_headers_length))
- .ok_or(IggyError::InvalidCommand)?;
- if bytes.len() < total_size {
- return Err(IggyError::InvalidCommand);
- }
-
- let payload_start = LEGACY_MESSAGE_HEADER_SIZE;
- let payload_end = payload_start + payload_length;
- let headers_end = payload_end + user_headers_length;
-
- Ok(Self {
- id: read_u128(bytes, 8)?,
- origin_timestamp: read_u64(bytes, 40)?,
- user_headers: &bytes[payload_end..headers_end],
- payload: &bytes[payload_start..payload_end],
- total_size,
- })
- }
-}
-
-/// Batch checksum v2: streaming `XxHash3_64` over the six batch header meta
-/// fields followed by each message's stored 8-byte checksum field in message
-/// order - NOT the message bodies.
-///
-/// Bodies are bound only transitively: each per-message checksum already covers
-/// `header[8..48] || payload || user_headers`, so hashing the checksum fields
-/// binds every body byte IFF a reader also re-verifies the per-message
-/// checksums. Stamp (produce) hashes `N * 8` bytes instead of the whole blob;
-/// validating decoders pay the one body pass as the per-message verify in
-/// [`verify_and_recompute_batch_checksum`], which hashes the checksum-field
-/// bytes in the same order so its recompute matches a compute here.
-///
-/// Assumes a well-formed blob whose frames tile exactly; every compute site
-/// builds the blob and satisfies this.
-fn calculate_batch_checksum(header: &SendMessagesHeader, blob: &[u8]) -> u64 {
- let mut hasher = XxHash3_64::new();
- write_batch_header_fields(&mut hasher, header);
- let batch = SendMessagesRef {
- header: *header,
- blob,
- };
- for framed in batch.iter_with_offsets() {
- hasher.write(&blob[framed.start..framed.start + 8]);
- }
- hasher.finish()
-}
-
-fn write_batch_header_fields(hasher: &mut XxHash3_64, header: &SendMessagesHeader) {
- hasher.write(&header.partition_id.to_le_bytes());
- hasher.write(&header.base_offset.to_le_bytes());
- hasher.write(&header.base_timestamp.to_le_bytes());
- hasher.write(&header.origin_timestamp.to_le_bytes());
- hasher.write(&header.batch_length.to_le_bytes());
- hasher.write(&header.message_count.to_le_bytes());
-}
-
-/// Verify every per-message checksum in `batch` and return the recomputed v2
-/// batch checksum (see [`calculate_batch_checksum`]) from a single frame walk.
-///
-/// The per-message pass is the equal-integrity half of v2: the batch value
-/// binds bodies only through the checksum fields, so a validating decode must
-/// re-verify each message here or body corruption that leaves the checksum
-/// field intact would pass. This is the one full-body pass a validating decode
-/// pays; the caller then compares the returned value against the stored
-/// `batch_checksum`.
-///
-/// # Errors
-///
-/// [`IggyError::InvalidMessageChecksum`] on the first per-message mismatch;
-/// [`IggyError::InvalidCommand`] if the frames do not tile `message_count`
-/// exactly.
-fn verify_and_recompute_batch_checksum(batch: &SendMessagesRef<'_>) -> Result<u64, IggyError> {
- let blob = batch.blob();
- let mut hasher = XxHash3_64::new();
- write_batch_header_fields(&mut hasher, &batch.header);
- let mut verified = 0u32;
- let mut covered = 0usize;
- for framed in batch.iter_with_offsets() {
- // Cover (`header[8..48] || payload || user_headers`) hashed raw from the
- // blob, byte-exact with the encoder's, so a flipped body byte fails even
- // when the stored checksum field is left intact.
- let stored = framed.message.header.checksum;
- let expected = XxHash3_64::oneshot(&blob[framed.start + 8..framed.end]);
- if expected != stored {
- return Err(IggyError::InvalidMessageChecksum(
- stored,
- expected,
- batch.header.base_offset + u64::from(framed.message.header.offset_delta),
- ));
- }
- hasher.write(&blob[framed.start..framed.start + 8]);
- verified += 1;
- covered = framed.end;
- }
- if verified != batch.message_count() || covered != blob.len() {
- return Err(IggyError::InvalidCommand);
- }
- Ok(hasher.finish())
-}
-
-/// The layout half of [`verify_and_recompute_batch_checksum`], without the hashing.
-///
-/// A caller that opts out of checksum verification still must not be handed a batch
-/// whose framing disagrees with its header, since it indexes by `batch_length` and
-/// would step into the next record. Walking the frames costs no hashing.
-fn validate_batch_layout(batch: &SendMessagesRef<'_>) -> Result<(), IggyError> {
- let mut framed = 0u32;
- let mut covered = 0usize;
- for message in batch.iter_with_offsets() {
- framed += 1;
- covered = message.end;
- }
- if framed != batch.message_count() || covered != batch.blob().len() {
- return Err(IggyError::InvalidCommand);
- }
- Ok(())
-}
-
-fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, IggyError> {
- bytes
- .get(offset..offset + 4)
- .and_then(|slice| slice.try_into().ok())
- .map(u32::from_le_bytes)
- .ok_or(IggyError::InvalidNumberEncoding)
-}
-
-fn read_u64(bytes: &[u8], offset: usize) -> Result<u64, IggyError> {
- bytes
- .get(offset..offset + 8)
- .and_then(|slice| slice.try_into().ok())
- .map(u64::from_le_bytes)
- .ok_or(IggyError::InvalidNumberEncoding)
-}
-
-fn read_u128(bytes: &[u8], offset: usize) -> Result<u128, IggyError> {
- bytes
- .get(offset..offset + 16)
- .and_then(|slice| slice.try_into().ok())
- .map(u128::from_le_bytes)
- .ok_or(IggyError::InvalidNumberEncoding)
-}
-
#[cfg(test)]
mod tests {
use super::*;
- use iggy_binary_protocol::{Command, Operation};
+ use iggy_binary_protocol::requests::messages::{RawMessage, SendMessagesEncoder};
+ use iggy_binary_protocol::{Command, Operation, WireIdentifier, WirePartitioning};
use iggy_common::Aes256GcmEncryptor;
+ use std::hash::Hasher;
fn aligned_prepare_bytes(size: u32) -> Owned<MESSAGE_ALIGN> {
let mut owned = Owned::<MESSAGE_ALIGN>::zeroed(std::mem::size_of::<PrepareHeader>());
@@ -1091,7 +699,7 @@ mod tests {
}
/// A checksum-consistent STAMPED `Prepare` carrying real per-message records,
- /// stamped at a non-zero `base_offset` / `base_timestamp` with a v2
+ /// stamped at a non-zero `base_offset` / `base_timestamp` with a
/// `batch_checksum` over the final header fields + per-message checksum fields.
fn valid_prepare_bytes() -> Owned<MESSAGE_ALIGN> {
let namespace = IggyNamespace::new(1, 1, 7);
@@ -1170,7 +778,7 @@ mod tests {
#[should_panic(expected = "must be at least 16-byte aligned")]
fn decode_prepare_slice_debug_asserts_on_misaligned_input() {
// `Vec<u8>` requests align=1; glibc returns a base that is a
- // multiple of 16, so `&buf[1..]` has offset 1 mod 16, eliably
+ // multiple of 16, so `&buf[1..]` has offset 1 mod 16, reliably
// misaligned.
let buf: Vec<u8> = vec![0u8; std::mem::size_of::<PrepareHeader>() + 1];
let misaligned = &buf[1..];
@@ -1257,8 +865,8 @@ mod tests {
}
#[test]
- fn batch_checksum_v2_pins_header_fields_then_message_checksum_fields() {
- // Formula pin for batch checksum v2: XxHash3-64 (default seed) streaming
+ fn batch_checksum_pins_header_fields_then_message_checksum_fields() {
+ // Formula pin for the batch checksum: XxHash3-64 (default seed) streaming
// over the six batch header meta fields (LE, in field order) then each
// message's stored 8-byte checksum field in message order - never the
// bodies. This reference walks the blob by the KNOWN input message sizes,
@@ -1284,7 +892,7 @@ mod tests {
for message in messages.iter() {
hasher.write(&owned.blob[frame_start..frame_start + 8]);
let user_headers = message.user_headers.as_deref().unwrap_or_default();
- frame_start += MESSAGE_HEADER_SIZE + message.payload.len() + user_headers.len();
+ frame_start += BATCH_MESSAGE_HEADER_SIZE + message.payload.len() + user_headers.len();
}
let reference = hasher.finish();
@@ -1295,17 +903,18 @@ mod tests {
);
assert_eq!(
owned.header.batch_checksum, reference,
- "v2 batch checksum must equal hash(6 header fields || per-message checksum fields)",
+ "batch checksum must equal hash(6 header fields || per-message checksum fields)",
);
}
#[test]
fn decode_batch_slice_rejects_body_corruption_with_intact_checksum_field() {
- // Equal-integrity: v2 binds bodies only through the per-message checksum
- // fields, so a flipped body byte that leaves the 8-byte checksum field
- // intact keeps the batch value matching. The validating decode must still
- // reject it via the per-message verify - the sole at-rest read-back check
- // (the poll disk walk) decodes through here.
+ // Equal-integrity: the batch value binds bodies only through the
+ // per-message checksum fields, so a flipped body byte that leaves the
+ // 8-byte checksum field intact keeps the batch value matching. The
+ // validating decode must still reject it via the per-message verify -
+ // the sole at-rest read-back check (the poll disk walk) decodes
+ // through here.
let namespace = IggyNamespace::new(1, 1, 7);
let owned =
SendMessagesOwned::from_messages(namespace, &sample_messages()).expect("build batch");
@@ -1317,7 +926,7 @@ mod tests {
// First payload byte sits right after the command header and the first
// message's 48B frame header, leaving that frame's checksum field intact.
- let payload_index = COMMAND_HEADER_SIZE + MESSAGE_HEADER_SIZE;
+ let payload_index = COMMAND_HEADER_SIZE + BATCH_MESSAGE_HEADER_SIZE;
body[payload_index] ^= 0xFF;
assert!(
matches!(
@@ -1335,7 +944,7 @@ mod tests {
let mut owned = prepare_with_messages(&sample_messages());
decode_prepare_slice(owned.as_slice()).expect("the clean prepare decodes");
- let payload_index = PREPARE_SPLIT_POINT + MESSAGE_HEADER_SIZE;
+ let payload_index = PREPARE_SPLIT_POINT + BATCH_MESSAGE_HEADER_SIZE;
owned.as_mut_slice()[payload_index] ^= 0xFF;
assert!(
matches!(
@@ -1350,29 +959,34 @@ mod tests {
);
}
- /// Legacy `SendMessages` request body: `[metadata_len=4][message_count]`
- /// then `count` skipped index slots, then the 64B-header legacy records.
- fn legacy_send_messages_body(messages: &IggyMessages) -> Vec<u8> {
- let count = messages.count();
- let mut body = Vec::new();
- body.extend_from_slice(&4u32.to_le_bytes());
- body.extend_from_slice(&count.to_le_bytes());
- body.extend_from_slice(&vec![0u8; count as usize * INDEX_SIZE]);
- for message in messages.iter() {
- let user_headers = message.user_headers.as_deref().unwrap_or_default();
- let mut header = [0u8; LEGACY_MESSAGE_HEADER_SIZE];
- header[8..24].copy_from_slice(&message.header.id.to_le_bytes());
- header[40..48].copy_from_slice(&message.header.origin_timestamp.to_le_bytes());
- header[48..52].copy_from_slice(&(user_headers.len() as u32).to_le_bytes());
- header[52..56].copy_from_slice(&(message.payload.len() as u32).to_le_bytes());
- body.extend_from_slice(&header);
- body.extend_from_slice(&message.payload);
- body.extend_from_slice(user_headers);
- }
- body
- }
-
- fn legacy_request_message(body: &[u8]) -> Message<RoutedRequestHeader> {
+ /// Wire-form `SendMessages` body built by the production client encoder:
+ /// `[metadata][256B batch header][blob]` with producer-computed checksums
+ /// and `partition_id = 0`.
+ fn wire_send_messages_body(messages: &IggyMessages) -> Vec<u8> {
+ let raw: Vec<RawMessage<'_>> = messages
+ .iter()
+ .map(|message| RawMessage {
+ id: message.header.id,
+ origin_timestamp: message.header.origin_timestamp,
+ headers: message.user_headers.as_deref(),
+ payload: &message.payload,
+ })
+ .collect();
+ let stream_id = WireIdentifier::numeric(1);
+ let topic_id = WireIdentifier::numeric(1);
+ let partitioning = WirePartitioning::Balanced;
+ let mut buf = BytesMut::with_capacity(SendMessagesEncoder::encoded_size(
+ &stream_id,
+ &topic_id,
+ &partitioning,
+ &raw,
+ ));
+ SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &raw)
+ .expect("wire body encodes");
+ buf.to_vec()
+ }
+
+ fn wire_request_message(body: &[u8]) -> Message<RoutedRequestHeader> {
let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total);
@@ -1388,11 +1002,11 @@ mod tests {
header.size = u32::try_from(total).expect("size fits u32");
}
buffer.as_mut_slice()[header_size..].copy_from_slice(body);
- Message::try_from(buffer).expect("legacy request message is valid")
+ Message::try_from(buffer).expect("wire request message is valid")
}
#[test]
- fn convert_request_message_rejects_empty_canonical_and_legacy_batches() {
+ fn convert_request_message_rejects_empty_batches() {
let namespace = IggyNamespace::new(1, 1, 3);
let messages = IggyMessages::with_capacity(0);
let canonical = SendMessagesOwned::from_messages(namespace, &messages)
@@ -1401,25 +1015,25 @@ mod tests {
canonical
.header
.encode_into(&mut canonical_body[..COMMAND_HEADER_SIZE]);
- let legacy_body = legacy_send_messages_body(&messages);
+ let wire_body = wire_send_messages_body(&messages);
for mode in [ChecksumMode::Compute, ChecksumMode::Skip] {
let canonical_result =
- convert_request_message(namespace, legacy_request_message(&canonical_body), mode);
+ convert_request_message(namespace, wire_request_message(&canonical_body), mode);
assert!(matches!(canonical_result, Err(IggyError::InvalidCommand)));
- let legacy_result =
- convert_request_message(namespace, legacy_request_message(&legacy_body), mode);
- assert!(matches!(legacy_result, Err(IggyError::InvalidCommand)));
+ let wire_result =
+ convert_request_message(namespace, wire_request_message(&wire_body), mode);
+ assert!(matches!(wire_result, Err(IggyError::InvalidCommand)));
}
}
#[test]
- fn convert_request_message_transcodes_legacy_to_canonical_bytes() {
- // Golden: the fused legacy transcode must emit the exact canonical batch
- // the native builder (`from_messages`) produces for the same messages -
- // command header + blob, byte for byte. Explicit non-zero ids keep it
- // deterministic (no `random_id` substitution).
+ fn convert_request_message_admits_wire_body_and_stamps_partition() {
+ // Golden: admitting the producer's wire body must yield the exact
+ // canonical batch the server-side builder (`from_messages`) produces
+ // for the same messages - command header + blob, byte for byte.
+ // Explicit non-zero ids keep it deterministic.
let namespace = IggyNamespace::new(1, 1, 3);
let messages = sample_messages();
@@ -1431,20 +1045,21 @@ mod tests {
.encode_into(&mut expected_body[..COMMAND_HEADER_SIZE]);
expected_body[COMMAND_HEADER_SIZE..].copy_from_slice(&owned.blob);
- let legacy = legacy_request_message(&legacy_send_messages_body(&messages));
- let converted = convert_request_message(namespace, legacy, ChecksumMode::Compute)
- .expect("legacy body transcodes");
+ let wire = wire_request_message(&wire_send_messages_body(&messages));
+ let converted = convert_request_message(namespace, wire, ChecksumMode::Compute)
+ .expect("wire body admits");
let header_size = std::mem::size_of::<RoutedRequestHeader>();
let actual_body = &converted.as_slice()[header_size..converted.header().size as usize];
assert_eq!(
actual_body, expected_body,
- "legacy transcode must be byte-identical to the canonical native batch",
+ "admitted wire body must be byte-identical to the server-built canonical batch",
);
- // And the emitted batch is self-consistent: it validates through the
+ // And the admitted batch is self-consistent: it validates through the
// batch-checksum decode and yields the original messages.
- let decoded = decode_batch_slice(actual_body).expect("transcoded batch checksum is valid");
+ let decoded = decode_batch_slice(actual_body).expect("admitted batch checksum is valid");
+ assert_eq!(decoded.header.partition_id, 3);
assert_eq!(decoded.message_count(), messages.count());
let payloads: Vec<&[u8]> = decoded.iter().map(|view| view.payload).collect();
assert_eq!(
@@ -1453,30 +1068,58 @@ mod tests {
);
}
+ #[test]
+ fn convert_request_message_rejects_tampered_wire_body() {
+ // A flipped payload byte invalidates the producer's per-message
+ // checksum; admission must refuse the batch instead of stamping it.
+ let namespace = IggyNamespace::new(1, 1, 3);
+ let mut body = wire_send_messages_body(&sample_messages());
+ let last = body.len() - 1;
+ body[last] ^= 0xFF;
+ let result =
+ convert_request_message(namespace, wire_request_message(&body), ChecksumMode::Skip);
+ assert!(matches!(result, Err(IggyError::InvalidMessageChecksum(..))));
+ }
+
+ #[test]
+ fn convert_request_message_rejects_metadata_count_mismatch() {
+ let namespace = IggyNamespace::new(1, 1, 3);
+ let mut body = wire_send_messages_body(&sample_messages());
+ // The metadata count is the 4 bytes right before the batch header:
+ // metadata = [stream][topic][partitioning][count].
+ let metadata_length =
+ u32::from_le_bytes(body[..4].try_into().expect("4-byte slice")) as usize;
+ let count_offset = 4 + metadata_length - 4;
+ body[count_offset..count_offset + 4].copy_from_slice(&9u32.to_le_bytes());
+ let result =
+ convert_request_message(namespace, wire_request_message(&body), ChecksumMode::Skip);
+ assert!(matches!(result, Err(IggyError::InvalidCommand)));
+ }
+
#[test]
fn convert_request_message_skip_leaves_batch_checksum_zero_until_stamp() {
- // The partition ingest path passes Skip: the transcoded batch must carry
- // a zero checksum (stamp fills it) and be otherwise byte-identical to the
- // Compute output - the flag toggles nothing but that one hash.
+ // The partition ingest path passes Skip: the admitted batch must carry
+ // a zero checksum (stamp fills it) and be otherwise byte-identical to
+ // the Compute output - the flag toggles nothing but that one hash.
let namespace = IggyNamespace::new(1, 1, 3);
let messages = sample_messages();
- let body = legacy_send_messages_body(&messages);
+ let body = wire_send_messages_body(&messages);
let header_size = std::mem::size_of::<RoutedRequestHeader>();
let computed = convert_request_message(
namespace,
- legacy_request_message(&body),
+ wire_request_message(&body),
ChecksumMode::Compute,
)
- .expect("compute transcode");
+ .expect("compute admission");
let skipped =
- convert_request_message(namespace, legacy_request_message(&body), ChecksumMode::Skip)
- .expect("skip transcode");
+ convert_request_message(namespace, wire_request_message(&body), ChecksumMode::Skip)
+ .expect("skip admission");
let computed_body = &computed.as_slice()[header_size..computed.header().size as usize];
let skipped_body = &skipped.as_slice()[header_size..skipped.header().size as usize];
- let skipped_header = SendMessagesHeader::decode(&skipped_body[..COMMAND_HEADER_SIZE])
+ let skipped_header = BatchHeader::decode(&skipped_body[..COMMAND_HEADER_SIZE])
.expect("decode skipped header");
assert_eq!(
skipped_header.batch_checksum, 0,
@@ -1506,9 +1149,9 @@ mod tests {
let messages = sample_messages();
let header_size = std::mem::size_of::<RoutedRequestHeader>();
- let legacy = legacy_request_message(&legacy_send_messages_body(&messages));
- let canonical = convert_request_message(namespace, legacy, ChecksumMode::Compute)
- .expect("pre-encrypt transcode");
+ let wire = wire_request_message(&wire_send_messages_body(&messages));
+ let canonical = convert_request_message(namespace, wire, ChecksumMode::Compute)
+ .expect("pre-encrypt admission");
let canonical_body = &canonical.as_slice()[header_size..canonical.header().size as usize];
decode_batch_slice(canonical_body).expect("encrypt input decode validates the checksum");
diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs
index ef398a404..3e7742ed1 100644
--- a/core/simulator/src/client.rs
+++ b/core/simulator/src/client.rs
@@ -70,12 +70,11 @@ pub struct SimClient {
/// [`SimClient::request_id_for`].
partition_counter: Cell<u64>,
/// Deterministic per-message id source for produced messages. The real SDK
- /// sends `id: 0` and lets the server mint a random UUID
- /// (`transcode_legacy_request` -> `random_id::get_uuid`); that
- /// mint is unseeded, so under the deterministic executor a produce's
- /// replicated body bytes (and their checksums) would differ run to run,
- /// silently breaking seeded replay. Stamping a deterministic id here keeps
- /// the body a pure function of the seed. See [`SimClient::next_message_id`].
+ /// mints a random UUID for a zero message id before encoding; that mint is
+ /// unseeded, so under the deterministic executor a produce's replicated
+ /// body bytes (and their checksums) would differ run to run, silently
+ /// breaking seeded replay. Stamping a deterministic id here keeps the body
+ /// a pure function of the seed. See [`SimClient::next_message_id`].
message_counter: Cell<u64>,
session: Cell<u64>,
}
@@ -512,12 +511,12 @@ impl SimClient {
self.build_request(Operation::DeletePersonalAccessToken, &wire.to_bytes())
}
- /// Build a `SendMessages` request in the legacy `SendMessagesEncoder` wire
- /// shape, byte-compatible with what the real SDK sends (`common` binary
- /// client). VSR clients resolve to an explicit partition before sending, so
- /// the sim always emits `WirePartitioning::PartitionId`: that is the shape
- /// the shell's `resolve_partition_request_namespace` decodes, and the raw
- /// path converts it to `SendMessages` via `transcode_legacy_request`.
+ /// Build a `SendMessages` request in the `SendMessagesEncoder` wire shape,
+ /// byte-compatible with what the real SDK sends (`common` binary client).
+ /// VSR clients resolve to an explicit partition before sending, so the sim
+ /// always emits `WirePartitioning::PartitionId`: that is the shape the
+ /// shell's `resolve_partition_request_namespace` decodes before admission
+ /// strips the metadata and stamps the batch.
///
/// # Panics
/// Panics if a group id exceeds `u32` or the request buffer is invalid.
@@ -548,7 +547,8 @@ impl SimClient {
let size = SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &raw);
let mut buf = BytesMut::with_capacity(size);
- SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &raw);
+ SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &raw)
+ .expect("simulator send batch encodes");
self.build_request_with_namespace(Operation::SendMessages, &buf, group)
}