You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2025/04/02 11:06:32 UTC
(iggy) branch zero-copy-no-batching updated: fix more tests
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/zero-copy-no-batching by this push:
new 0a818763 fix more tests
0a818763 is described below
commit 0a818763f30fa4b0ae853b8b0cd14e357eaf8c22
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Apr 2 13:06:23 2025 +0200
fix more tests
---
integration/tests/examples/mod.rs | 2 +-
integration/tests/mod.rs | 2 +-
...with_single_client_polling_messages_scenario.rs | 8 +-
.../server/scenarios/message_size_scenario.rs | 38 +++--
integration/tests/server/scenarios/mod.rs | 2 +-
.../scenarios/stream_size_validation_scenario.rs | 2 +-
.../tests/server/scenarios/system_scenario.rs | 2 +-
integration/tests/streaming/segment.rs | 6 +-
sdk/src/error.rs | 4 +-
sdk/src/messages/mod.rs | 2 +-
sdk/src/messages/polled_messages.rs | 1 -
sdk/src/messages/send_messages.rs | 3 +-
sdk/src/models/messaging/indexes.rs | 5 +
sdk/src/models/messaging/messages_batch.rs | 7 +-
sdk/src/prelude.rs | 1 +
.../handlers/messages/poll_messages_handler.rs | 13 +-
.../handlers/messages/send_messages_handler.rs | 15 +-
.../src/binary/handlers/system/get_me_handler.rs | 1 -
server/src/binary/handlers/utils.rs | 15 +-
server/src/quic/quic_sender.rs | 6 +-
server/src/streaming/partitions/messages.rs | 2 +-
.../src/streaming/segments/indexes/index_writer.rs | 2 +-
.../src/streaming/segments/indexes/indexes_mut.rs | 23 +--
.../src/streaming/segments/messages_accumulator.rs | 2 +
server/src/streaming/segments/reading_messages.rs | 19 ++-
server/src/streaming/segments/segment.rs | 4 +-
.../streaming/segments/types/messages_batch_mut.rs | 156 +++++++++++++++++++++
server/src/streaming/segments/writing_messages.rs | 2 +
server/src/tcp/sender.rs | 24 +++-
29 files changed, 285 insertions(+), 84 deletions(-)
diff --git a/integration/tests/examples/mod.rs b/integration/tests/examples/mod.rs
index 3f9e8dd0..7d2aa010 100644
--- a/integration/tests/examples/mod.rs
+++ b/integration/tests/examples/mod.rs
@@ -174,7 +174,7 @@ impl IggyExampleTest<'_> {
let consumer_handle = tokio::spawn(async move {
let consumer_assert = consumer_cmd
- .timeout(Duration::from_secs(2))
+ .timeout(Duration::from_secs(10))
.assert()
.success();
let consumer_output = consumer_assert.get_output();
diff --git a/integration/tests/mod.rs b/integration/tests/mod.rs
index dfba2f1c..46821879 100644
--- a/integration/tests/mod.rs
+++ b/integration/tests/mod.rs
@@ -33,7 +33,7 @@ fn setup() {
let mut logger = env_logger::builder();
logger.is_test(true);
- logger.filter(None, log::LevelFilter::Trace);
+ logger.filter(None, log::LevelFilter::Trace); // todo
logger.target(env_logger::Target::Pipe(Box::new(LogWriter(log_buffer))));
logger.format(move |buf, record| {
let timestamp = chrono::Local::now().format("%Y-%m-%d %H:%M:%S%.6f");
diff --git a/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs b/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
index 3136c10f..19878f4b 100644
--- a/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
+++ b/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs
@@ -143,7 +143,7 @@ async fn execute_using_none_key(client: &IggyClient) {
let mut partition_id = 1;
let mut offset = 0;
let mut entity_id = 1;
- for _ in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
+ for i in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
let polled_messages = client
.poll_messages(
&Identifier::numeric(STREAM_ID).unwrap(),
@@ -157,7 +157,11 @@ async fn execute_using_none_key(client: &IggyClient) {
.await
.unwrap();
- assert_eq!(polled_messages.messages.len(), 1);
+ assert_eq!(
+ polled_messages.messages.len(),
+ 1,
+ "polled messages count is not 1 at iteration {i}"
+ );
let message = &polled_messages.messages[0];
assert_eq!(message.header.offset, offset);
let payload = from_utf8(&message.payload).unwrap();
diff --git a/integration/tests/server/scenarios/message_size_scenario.rs b/integration/tests/server/scenarios/message_size_scenario.rs
index bfbb5700..c0d9f909 100644
--- a/integration/tests/server/scenarios/message_size_scenario.rs
+++ b/integration/tests/server/scenarios/message_size_scenario.rs
@@ -10,7 +10,7 @@ const STREAM_NAME: &str = "test-stream";
const TOPIC_NAME: &str = "test-topic";
const PARTITIONS_COUNT: u32 = 3;
const PARTITION_ID: u32 = 1;
-const MAX_HEADER_SIZE: usize = 255;
+const MAX_SINGLE_HEADER_SIZE: usize = 200;
enum MessageToSend {
NoMessage,
@@ -62,7 +62,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
send_message_and_check_result(
&client,
MessageToSend::OfSizeWithHeaders(100_000, 10_000_001),
- Err(IggyError::TooBigMessagePayload(100_000, 10_000_001)),
+ Err(IggyError::TooBigMessagePayload),
)
.await;
@@ -164,23 +164,35 @@ fn create_string_of_size(size: usize) -> String {
"x".repeat(size)
}
-fn create_message_header_of_size(size: usize) -> HashMap<HeaderKey, HeaderValue> {
+fn create_message_header_of_size(target_size: usize) -> HashMap<HeaderKey, HeaderValue> {
let mut headers = HashMap::new();
let mut header_id = 1;
- let mut size = size;
+ let mut current_size = 0;
- while size > 0 {
- let header_size = if size > MAX_HEADER_SIZE {
- MAX_HEADER_SIZE
+ while current_size < target_size {
+ let remaining_size = target_size - current_size;
+
+ let key_str = format!("header-{header_id}");
+ let key_overhead = 4; // 4 bytes for key length
+ let value_overhead = 5; // 1 byte for type + 4 bytes for value length
+ let total_overhead = key_overhead + key_str.len() + value_overhead;
+
+ let value_size = if remaining_size <= total_overhead {
+ break;
+ } else if remaining_size - total_overhead > MAX_SINGLE_HEADER_SIZE {
+ MAX_SINGLE_HEADER_SIZE
} else {
- size
+ remaining_size - total_overhead
};
- headers.insert(
- HeaderKey::new(format!("header-{header_id}").as_str()).unwrap(),
- HeaderValue::from_str(create_string_of_size(header_size).as_str()).unwrap(),
- );
+
+ let key = HeaderKey::new(key_str.as_str()).unwrap();
+ let value = HeaderValue::from_str(create_string_of_size(value_size).as_str()).unwrap();
+
+ let actual_header_size = 4 + key_str.len() + 1 + 4 + value_size;
+ current_size += actual_header_size;
+
+ headers.insert(key, value);
header_id += 1;
- size -= header_size;
}
headers
diff --git a/integration/tests/server/scenarios/mod.rs b/integration/tests/server/scenarios/mod.rs
index d37a81d7..eafe2c46 100644
--- a/integration/tests/server/scenarios/mod.rs
+++ b/integration/tests/server/scenarios/mod.rs
@@ -28,7 +28,7 @@ const USERNAME_2: &str = "user2";
const USERNAME_3: &str = "user3";
const CONSUMER_ID: u32 = 1;
const CONSUMER_KIND: ConsumerKind = ConsumerKind::Consumer;
-const MESSAGES_COUNT: u32 = 1000;
+const MESSAGES_COUNT: u32 = 1337;
async fn create_client(client_factory: &dyn ClientFactory) -> IggyClient {
let client = client_factory.create_client().await;
diff --git a/integration/tests/server/scenarios/stream_size_validation_scenario.rs b/integration/tests/server/scenarios/stream_size_validation_scenario.rs
index 266544bb..27a62a61 100644
--- a/integration/tests/server/scenarios/stream_size_validation_scenario.rs
+++ b/integration/tests/server/scenarios/stream_size_validation_scenario.rs
@@ -12,7 +12,7 @@ const T1_NAME: &str = "test-topic-1";
const S2_NAME: &str = "test-stream-2";
const T2_NAME: &str = "test-topic-2";
const MESSAGE_PAYLOAD_SIZE_BYTES: u64 = 57;
-const MSG_SIZE: u64 = 16 + 8 + 8 + 4 + 4 + 4 + 1 + MESSAGE_PAYLOAD_SIZE_BYTES; // number of bytes in a single message
+const MSG_SIZE: u64 = IGGY_MESSAGE_HEADER_SIZE as u64 + MESSAGE_PAYLOAD_SIZE_BYTES; // number of bytes in a single message
const MSGS_COUNT: u64 = 117; // number of messages in a single topic after one pass of appending
const MSGS_SIZE: u64 = MSG_SIZE * MSGS_COUNT; // number of bytes in a single topic after one pass of appending
diff --git a/integration/tests/server/scenarios/system_scenario.rs b/integration/tests/server/scenarios/system_scenario.rs
index f6d0f74f..e027d0e0 100644
--- a/integration/tests/server/scenarios/system_scenario.rs
+++ b/integration/tests/server/scenarios/system_scenario.rs
@@ -273,7 +273,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
assert_eq!(topic.name, TOPIC_NAME);
assert_eq!(topic.partitions_count, PARTITIONS_COUNT);
assert_eq!(topic.partitions.len(), PARTITIONS_COUNT as usize);
- assert_eq!(topic.size, 66890);
+ assert_eq!(topic.size, 89806);
assert_eq!(topic.messages_count, MESSAGES_COUNT as u64);
let topic_partition = topic.partitions.get((PARTITION_ID - 1) as usize).unwrap();
assert_eq!(topic_partition.id, PARTITION_ID);
diff --git a/integration/tests/streaming/segment.rs b/integration/tests/streaming/segment.rs
index 91e64cf2..42549726 100644
--- a/integration/tests/streaming/segment.rs
+++ b/integration/tests/streaming/segment.rs
@@ -317,9 +317,11 @@ async fn given_all_expired_messages_segment_should_be_expired() {
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
segment.append_batch(0, batch).await.unwrap();
segment.persist_messages(None).await.unwrap();
+ let not_expired_ts = IggyTimestamp::now();
+ let expired_ts = not_expired_ts + IggyDuration::from(message_expiry_us + 1);
- let is_expired = segment.is_expired(IggyTimestamp::now()).await;
- assert!(is_expired);
+ assert!(segment.is_expired(expired_ts).await);
+ assert!(!segment.is_expired(not_expired_ts).await);
}
#[tokio::test]
diff --git a/sdk/src/error.rs b/sdk/src/error.rs
index 84c21742..72bd11c6 100644
--- a/sdk/src/error.rs
+++ b/sdk/src/error.rs
@@ -312,8 +312,8 @@ pub enum IggyError {
CannotReadMessageLength = 4020,
#[error("Cannot save messages to segment")]
CannotReadMessagePayload = 4021,
- #[error("Too big message payload: {0}, must be <= {1}")]
- TooBigMessagePayload(u32, u32) = 4022,
+ #[error("Too big message payload")]
+ TooBigMessagePayload = 4022,
#[error("Too many messages")]
TooManyMessages = 4023,
#[error("Empty message payload")]
diff --git a/sdk/src/messages/mod.rs b/sdk/src/messages/mod.rs
index 09c543bf..ae88365d 100644
--- a/sdk/src/messages/mod.rs
+++ b/sdk/src/messages/mod.rs
@@ -7,7 +7,7 @@ mod polling_kind;
mod polling_strategy;
mod send_messages;
-pub const MAX_HEADERS_SIZE: u32 = 100 * 1000;
+pub const MAX_USER_HEADERS_SIZE: u32 = 100 * 1000;
pub const MAX_PAYLOAD_SIZE: u32 = 10 * 1000 * 1000;
pub use flush_unsaved_buffer::FlushUnsavedBuffer;
diff --git a/sdk/src/messages/polled_messages.rs b/sdk/src/messages/polled_messages.rs
index ca848a6e..4afb35a2 100644
--- a/sdk/src/messages/polled_messages.rs
+++ b/sdk/src/messages/polled_messages.rs
@@ -72,7 +72,6 @@ fn messages_from_bytes_and_count(buffer: Bytes, count: u32) -> Result<Vec<IggyMe
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 as usize > buf_len {
break;
diff --git a/sdk/src/messages/send_messages.rs b/sdk/src/messages/send_messages.rs
index 0fc99d64..e329555e 100644
--- a/sdk/src/messages/send_messages.rs
+++ b/sdk/src/messages/send_messages.rs
@@ -125,8 +125,7 @@ impl Validatable<IggyError> for SendMessages {
return Err(IggyError::InvalidKeyValueLength);
}
- // TODO(hubcio): fix validation
- // self.batch.validate()?;
+ self.batch.validate()?;
Ok(())
}
diff --git a/sdk/src/models/messaging/indexes.rs b/sdk/src/models/messaging/indexes.rs
index 500b1d2f..0ac98254 100644
--- a/sdk/src/models/messaging/indexes.rs
+++ b/sdk/src/models/messaging/indexes.rs
@@ -38,6 +38,11 @@ impl IggyIndexes {
self.buffer.len() as u32 / INDEX_SIZE as u32
}
+ /// Gets the size of the buffer in bytes
+ pub fn size(&self) -> u32 {
+ self.buffer.len() as u32
+ }
+
/// Gets the size of all indexes messages
pub fn messages_size(&self) -> u32 {
self.last_position() - self.base_position
diff --git a/sdk/src/models/messaging/messages_batch.rs b/sdk/src/models/messaging/messages_batch.rs
index 52892063..3ff535bc 100644
--- a/sdk/src/models/messaging/messages_batch.rs
+++ b/sdk/src/models/messaging/messages_batch.rs
@@ -413,7 +413,7 @@ impl Validatable<IggyError> for IggyMessagesBatch {
}
let indexes_count = self.indexes.count();
- let indexes_size = self.indexes.messages_size();
+ let indexes_size = self.indexes.size();
if indexes_size % INDEX_SIZE as u32 != 0 {
tracing::error!(
@@ -462,10 +462,7 @@ impl Validatable<IggyError> for IggyMessagesBatch {
message.payload().len(),
MAX_PAYLOAD_SIZE
);
- return Err(IggyError::TooBigMessagePayload(
- message.payload().len() as u32,
- MAX_PAYLOAD_SIZE,
- ));
+ return Err(IggyError::TooBigMessagePayload);
}
messages_size += message.size();
diff --git a/sdk/src/prelude.rs b/sdk/src/prelude.rs
index 15b30a82..743a941d 100644
--- a/sdk/src/prelude.rs
+++ b/sdk/src/prelude.rs
@@ -31,6 +31,7 @@ pub use crate::messages::{
FlushUnsavedBuffer, Partitioning, PollMessages, PolledMessages, PollingKind, PollingStrategy,
SendMessages,
};
+pub use crate::messages::{MAX_PAYLOAD_SIZE, MAX_USER_HEADERS_SIZE};
pub use crate::models::consumer_group::ConsumerGroupDetails;
pub use crate::models::messaging::{
HeaderKey, HeaderValue, IggyMessage, IggyMessageHeader, IggyMessageHeaderView, IggyMessageView,
diff --git a/server/src/binary/handlers/messages/poll_messages_handler.rs b/server/src/binary/handlers/messages/poll_messages_handler.rs
index 95742586..1214c698 100644
--- a/server/src/binary/handlers/messages/poll_messages_handler.rs
+++ b/server/src/binary/handlers/messages/poll_messages_handler.rs
@@ -9,7 +9,7 @@ use anyhow::Result;
use error_set::ErrContext;
use iggy::prelude::*;
use std::io::IoSlice;
-use tracing::debug;
+use tracing::{debug, trace};
#[derive(Debug)]
pub struct IggyPollMetadata {
@@ -63,7 +63,8 @@ impl ServerCommandHandler for PollMessages {
// long enough" errors while optimizing transmission by using larger chunks.
// 4 bytes for partition_id + 8 bytes for current_offset + 4 bytes for messages_count + size of all batches.
- let response_length = (4 + 8 + 4 + messages.size()).to_le_bytes();
+ let response_length = 4 + 8 + 4 + messages.size();
+ let response_length_bytes = response_length.to_le_bytes();
let partition_id = metadata.partition_id.to_le_bytes();
let current_offset = metadata.current_offset.to_le_bytes();
@@ -76,8 +77,14 @@ impl ServerCommandHandler for PollMessages {
io_slices.extend(messages.iter().map(|m| IoSlice::new(m)));
+ trace!(
+ "Sending {} messages to client ({} bytes) to client",
+ messages.count(),
+ response_length
+ );
+
sender
- .send_ok_response_vectored(&response_length, io_slices)
+ .send_ok_response_vectored(&response_length_bytes, io_slices)
.await?;
Ok(())
}
diff --git a/server/src/binary/handlers/messages/send_messages_handler.rs b/server/src/binary/handlers/messages/send_messages_handler.rs
index 5368ca10..6a7a933b 100644
--- a/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -56,26 +56,15 @@ impl ServerCommandHandler for SendMessages {
let mut indexes_and_messages = buffer.split();
- tracing::error!(
- "indexes_and_messages len {}, messages_count {}",
- indexes_and_messages.len(),
- messages_count
- );
-
let messages = indexes_and_messages.split_off(messages_count * INDEX_SIZE);
let indexes = indexes_and_messages;
let indexes = IggyIndexesMut::from_bytes(indexes);
- tracing::error!(
- "hubcio messages part len {}, indexes size {}, indexes count {}",
- messages.len(),
- indexes.size(),
- indexes.count(),
- );
-
let batch = IggyMessagesBatchMut::from_indexes_and_messages(indexes, messages);
+ batch.validate()?;
+
let system = system.read().await;
system
.append_messages(session, &stream_id, &topic_id, &partitioning, batch, None)
diff --git a/server/src/binary/handlers/system/get_me_handler.rs b/server/src/binary/handlers/system/get_me_handler.rs
index a52b137a..b0e44d75 100644
--- a/server/src/binary/handlers/system/get_me_handler.rs
+++ b/server/src/binary/handlers/system/get_me_handler.rs
@@ -9,7 +9,6 @@ use error_set::ErrContext;
use iggy::error::IggyError;
use iggy::locking::IggySharedMutFn;
use iggy::system::get_me::GetMe;
-use tracing::debug;
impl ServerCommandHandler for GetMe {
fn code(&self) -> u32 {
diff --git a/server/src/binary/handlers/utils.rs b/server/src/binary/handlers/utils.rs
index 710e4a40..bcb3289f 100644
--- a/server/src/binary/handlers/utils.rs
+++ b/server/src/binary/handlers/utils.rs
@@ -1,5 +1,5 @@
use crate::binary::{command::ServerCommand, sender::SenderKind};
-use bytes::{BufMut, BytesMut};
+use bytes::BytesMut;
use iggy::error::IggyError;
pub async fn receive_and_validate(
@@ -8,19 +8,12 @@ pub async fn receive_and_validate(
length: u32,
) -> Result<ServerCommand, IggyError> {
let mut buffer = BytesMut::with_capacity(length as usize);
-
- unsafe {
- buffer.set_len(length as usize);
- }
- tracing::error!("length: {length}");
- // buffer.put_bytes(0, length as usize);
- tracing::error!("after put");
if length > 0 {
- tracing::error!("reading");
+ unsafe {
+ buffer.set_len(length as usize);
+ }
sender.read(&mut buffer).await?;
- tracing::error!("after read");
}
-
let command = ServerCommand::from_code_and_payload(code, buffer.freeze())?;
command.validate()?;
Ok(command)
diff --git a/server/src/quic/quic_sender.rs b/server/src/quic/quic_sender.rs
index aee3253f..c3c4d7f4 100644
--- a/server/src/quic/quic_sender.rs
+++ b/server/src/quic/quic_sender.rs
@@ -95,7 +95,11 @@ impl Sender for QuicSender {
impl QuicSender {
async fn send_response(&mut self, status: &[u8], payload: &[u8]) -> Result<(), IggyError> {
- debug!("Sending response with status: {:?}...", status);
+ debug!(
+ "Sending response of len: {} with status: {:?}...",
+ payload.len(),
+ status
+ );
let length = (payload.len() as u32).to_le_bytes();
self.send
.write_all(&[status, &length, payload].as_slice().concat())
diff --git a/server/src/streaming/partitions/messages.rs b/server/src/streaming/partitions/messages.rs
index 9432561f..c5f1423e 100644
--- a/server/src/streaming/partitions/messages.rs
+++ b/server/src/streaming/partitions/messages.rs
@@ -116,7 +116,7 @@ impl Partition {
let offset = consumer_offset.offset + 1;
trace!(
- "Getting next messages for {} for partition: {} from offset: {}...",
+ "Getting next messages for consumer id: {} for partition: {} from offset: {}...",
consumer_id,
self.partition_id,
offset
diff --git a/server/src/streaming/segments/indexes/index_writer.rs b/server/src/streaming/segments/indexes/index_writer.rs
index a4e0eb43..d3094a9f 100644
--- a/server/src/streaming/segments/indexes/index_writer.rs
+++ b/server/src/streaming/segments/indexes/index_writer.rs
@@ -87,7 +87,7 @@ impl IndexWriter {
self.index_size_bytes
.fetch_add(indexes.len() as u64, Ordering::Release);
- trace!("XXXX Saved {count} indexes to file: {}", self.file_path);
+ trace!("Saved {count} indexes to file: {}", self.file_path);
Ok(())
}
diff --git a/server/src/streaming/segments/indexes/indexes_mut.rs b/server/src/streaming/segments/indexes/indexes_mut.rs
index 91bf98b2..7b1dd6d1 100644
--- a/server/src/streaming/segments/indexes/indexes_mut.rs
+++ b/server/src/streaming/segments/indexes/indexes_mut.rs
@@ -28,6 +28,18 @@ impl IggyIndexesMut {
}
}
+ /// Gets the size of all indexes messages
+ pub fn messages_size(&self) -> u32 {
+ self.last_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) -> Self {
Self {
@@ -64,8 +76,8 @@ impl IggyIndexesMut {
}
/// Gets the size of the buffer in bytes
- pub fn size(&self) -> usize {
- self.buffer.len()
+ pub fn size(&self) -> u32 {
+ self.buffer.len() as u32
}
/// Gets a view of the index at the specified position
@@ -123,10 +135,6 @@ impl IggyIndexesMut {
let first_idx = self.get(0)?;
if timestamp <= first_idx.timestamp() {
- tracing::trace!(
- "Requested timestamp {} is less than any available",
- timestamp
- );
return Some(first_idx);
}
@@ -199,8 +207,7 @@ impl IggyIndexesMut {
let end_byte = end_pos as usize * INDEX_SIZE;
let slice = BytesMut::from(&self.buffer[start_byte..end_byte]);
- let i = relative_start_offset.saturating_sub(1);
- if i == 0 {
+ if relative_start_offset == 0 {
Some(IggyIndexes::new(slice.freeze(), 0))
} else {
let base_position = self.get(relative_start_offset - 1).unwrap().position();
diff --git a/server/src/streaming/segments/messages_accumulator.rs b/server/src/streaming/segments/messages_accumulator.rs
index 33e91503..397ef73e 100644
--- a/server/src/streaming/segments/messages_accumulator.rs
+++ b/server/src/streaming/segments/messages_accumulator.rs
@@ -108,6 +108,8 @@ impl MessagesAccumulator {
///
/// A batch set containing the requested messages
pub fn get_messages_by_offset(&self, start_offset: u64, count: u32) -> IggyMessagesBatchSet {
+ trace!("Getting {count} messages from accumulator by offset {start_offset}, current_offset: {}, current_position: {}",
+ self.current_offset, self.current_position);
self.batches.get_by_offset(start_offset, count)
}
diff --git a/server/src/streaming/segments/reading_messages.rs b/server/src/streaming/segments/reading_messages.rs
index 8cdc94c5..b23b8323 100644
--- a/server/src/streaming/segments/reading_messages.rs
+++ b/server/src/streaming/segments/reading_messages.rs
@@ -9,12 +9,18 @@ use tracing::trace;
const COMPONENT: &str = "STREAMING_SEGMENT";
+// TODO: test when offset > u32
+
impl Segment {
pub fn get_messages_size(&self) -> IggyByteSize {
IggyByteSize::from(self.messages_size.load(Ordering::Relaxed))
}
pub fn get_messages_count(&self) -> u32 {
+ if self.get_messages_size() == 0 {
+ return 0;
+ }
+
(self.end_offset - self.start_offset + 1) as u32
}
@@ -243,9 +249,9 @@ impl Segment {
count: u32,
) -> Result<IggyMessagesBatchSet, IggyError> {
tracing::trace!(
- "Loading {count} messages from disk, start_offset: {start_offset}, current_offset: {}...",
- self.end_offset
- );
+ "Loading {count} messages from disk, start_offset: {start_offset}, end_offset: {}...",
+ self.end_offset
+ );
let relative_start_offset = (start_offset - self.start_offset) as u32;
let indexes_to_read = self
@@ -267,7 +273,12 @@ impl Segment {
format!("Failed to load messages from segment file: {self}. {error}")
})?;
- tracing::trace!("Loaded {} messages from disk", msgs.count());
+ tracing::trace!(
+ "Loaded {} messages ({} bytes) from disk (requested {count} messages), start_offset: {start_offset}, end_offset: {}",
+ msgs.count(),
+ msgs.size(),
+ self.end_offset
+ );
Ok(msgs)
}
diff --git a/server/src/streaming/segments/segment.rs b/server/src/streaming/segments/segment.rs
index 2567271d..319cbe67 100644
--- a/server/src/streaming/segments/segment.rs
+++ b/server/src/streaming/segments/segment.rs
@@ -151,7 +151,7 @@ impl Segment {
let messages_count = self.get_messages_count() as u64;
info!(
- "Loaded segment with log file of size {} ({} messages) for start offset {}, current offset: {}, and partition with ID: {} for topic with ID: {} and stream with ID: {}.",
+ "Loaded segment with log file of size {} ({} messages) for start offset {}, end offset: {}, and partition with ID: {} for topic with ID: {} and stream with ID: {}.",
IggyByteSize::from(log_size_bytes),
messages_count,
self.start_offset,
@@ -387,7 +387,7 @@ impl std::fmt::Display for Segment {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "Segment {{ stream_id: {}, topic_id: {}, partition_id: {}, start_offset: {}, current_offset: {}, size_bytes: {}, last_index_position: {}, max_size_bytes: {}, closed: {} }}",
+ "Segment {{ stream_id: {}, topic_id: {}, partition_id: {}, start_offset: {}, end_offset: {}, size_bytes: {}, last_index_position: {}, max_size_bytes: {}, closed: {} }}",
self.stream_id,
self.topic_id,
self.partition_id,
diff --git a/server/src/streaming/segments/types/messages_batch_mut.rs b/server/src/streaming/segments/types/messages_batch_mut.rs
index 7bb16e2a..e2c3a2a5 100644
--- a/server/src/streaming/segments/types/messages_batch_mut.rs
+++ b/server/src/streaming/segments/types/messages_batch_mut.rs
@@ -2,6 +2,7 @@ use super::message_view_mut::IggyMessageViewMutIterator;
use crate::streaming::segments::indexes::IggyIndexesMut;
use crate::streaming::utils::random_id;
use bytes::{BufMut, BytesMut};
+use iggy::messages::MAX_PAYLOAD_SIZE;
use iggy::models::messaging::{IggyMessagesBatch, INDEX_SIZE};
use iggy::prelude::*;
use iggy::utils::timestamp::IggyTimestamp;
@@ -182,6 +183,161 @@ impl IggyMessagesBatchMut {
pub fn decompose(self) -> (IggyIndexesMut, BytesMut) {
(self.indexes, self.messages)
}
+
+ /// Helper method to read a position (u32) from the byte array at the given index
+ fn position_at(&self, position_index: u32) -> u32 {
+ if let Some(index) = self.indexes.get(position_index) {
+ index.position()
+ } else {
+ 0
+ }
+ }
+
+ /// 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;
+ }
+
+ // TODO: refactor this
+ let start_position = if index == 0 {
+ 0
+ } else {
+ self.position_at(index as u32 - 1) as usize
+ };
+
+ let end_position = if index == self.count() as usize - 1 {
+ self.messages.len()
+ } else {
+ self.position_at(index as u32) as usize
+ };
+
+ debug_assert!(
+ start_position <= self.messages.len(),
+ "Start message position {} exceeds buffer length {}, index: {}, self.count: {}",
+ start_position,
+ self.messages.len(),
+ index,
+ self.count(),
+ );
+ debug_assert!(
+ end_position <= self.messages.len(),
+ "End message position {} exceeds buffer length {}, index: {}, self.count: {}",
+ end_position,
+ self.messages.len(),
+ index,
+ self.count(),
+ );
+
+ Some(IggyMessageView::new(
+ &self.messages[start_position..end_position],
+ ))
+ }
+}
+
+impl Validatable<IggyError> for IggyMessagesBatchMut {
+ fn validate(&self) -> Result<(), IggyError> {
+ if self.is_empty() {
+ return Err(IggyError::InvalidMessagesCount);
+ }
+
+ let indexes_count = self.indexes.count();
+ let indexes_size = self.indexes.size();
+
+ if indexes_size % INDEX_SIZE as u32 != 0 {
+ tracing::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() {
+ tracing::error!(
+ "Indexes count {} does not match messages count {}",
+ indexes_count,
+ self.count()
+ );
+ return Err(IggyError::InvalidIndexesCount(indexes_count, self.count()));
+ }
+
+ let mut messages_count = 0;
+ let mut messages_size = 0;
+
+ for i in 0..self.count() {
+ if let Some(index_view) = self.indexes.get(i) {
+ if index_view.offset() != 0 {
+ tracing::error!("Non-zero offset {} at index: {}", index_view.offset(), i);
+ return Err(IggyError::NonZeroOffset(index_view.offset() as u64, i));
+ }
+ if index_view.timestamp() != 0 {
+ tracing::error!(
+ "Non-zero timestamp {} at index: {}",
+ index_view.timestamp(),
+ i
+ );
+ return Err(IggyError::NonZeroTimestamp(index_view.timestamp(), i));
+ }
+ } else {
+ tracing::error!("Index {} is missing", i);
+ return Err(IggyError::MissingIndex(i));
+ }
+
+ if let Some(message) = self.get(i as usize) {
+ if message.payload().len() as u32 > MAX_PAYLOAD_SIZE {
+ tracing::error!(
+ "Message payload size {} B exceeds maximum payload size {} B",
+ message.payload().len(),
+ MAX_PAYLOAD_SIZE
+ );
+ return Err(IggyError::TooBigMessagePayload);
+ }
+
+ if let Some(user_headers) = message.user_headers() {
+ tracing::error!("message user headers size {}", user_headers.len());
+ if user_headers.len() as u32 > MAX_USER_HEADERS_SIZE {
+ tracing::error!(
+ "Message user headers size {} B exceeds maximum size {} B",
+ user_headers.len(),
+ MAX_USER_HEADERS_SIZE
+ );
+ return Err(IggyError::TooBigHeadersPayload);
+ }
+ }
+
+ messages_size += message.size();
+ messages_count += 1;
+ } else {
+ tracing::error!("Missing index {}", i);
+ return Err(IggyError::MissingIndex(i));
+ }
+ }
+
+ if indexes_count != messages_count {
+ tracing::error!(
+ "Indexes count {} does not match messages count {}",
+ indexes_count,
+ messages_count
+ );
+ return Err(IggyError::InvalidMessagesCount);
+ }
+
+ if messages_size != self.messages.len() as u32 {
+ tracing::error!(
+ "Messages size {} does not match messages buffer size {}",
+ messages_size,
+ self.messages.len() as u64
+ );
+ return Err(IggyError::InvalidMessagesSize(
+ messages_size,
+ self.messages.len() as u32,
+ ));
+ }
+
+ Ok(())
+ }
}
impl Deref for IggyMessagesBatchMut {
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index ed768104..8599ee79 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -129,6 +129,8 @@ impl Segment {
.fetch_add(messages_count, Ordering::SeqCst);
self.messages_count_of_parent_partition
.fetch_add(messages_count, Ordering::SeqCst);
+ self.messages_size
+ .fetch_add(messages_size, Ordering::SeqCst);
}
async fn check_and_handle_segment_full(&mut self) -> Result<(), IggyError> {
diff --git a/server/src/tcp/sender.rs b/server/src/tcp/sender.rs
index 41db1aec..f86c9b97 100644
--- a/server/src/tcp/sender.rs
+++ b/server/src/tcp/sender.rs
@@ -65,7 +65,11 @@ pub(crate) async fn send_response<T>(
where
T: AsyncRead + AsyncWrite + Unpin,
{
- debug!("Sending response with status: {:?}...", status);
+ debug!(
+ "Sending response of len: {} with status: {:?}...",
+ payload.len(),
+ status
+ );
let length = (payload.len() as u32).to_le_bytes();
stream
.write_all(&[status, &length, payload].as_slice().concat())
@@ -84,13 +88,21 @@ pub(crate) async fn send_response_vectored<T>(
where
T: AsyncReadExt + AsyncWriteExt + Unpin,
{
- debug!("Sending response with status: {:?}...", status);
+ debug!(
+ "Sending vectored response of len: {} with status: {:?}...",
+ slices.len(),
+ status
+ );
let prefix = [IoSlice::new(status), IoSlice::new(length)];
slices.splice(0..0, prefix);
- stream
- .write_vectored(&slices)
- .await
- .map_err(|_| IggyError::TcpError)?;
+ let mut slice_refs = slices.as_mut_slice();
+ while !slice_refs.is_empty() {
+ let bytes_written = stream
+ .write_vectored(slice_refs)
+ .await
+ .map_err(|_| IggyError::TcpError)?;
+ IoSlice::advance_slices(&mut slice_refs, bytes_written);
+ }
debug!("Sent response with status: {:?}", status);
Ok(())
}