fix more tests
diff --git a/Cargo.toml b/Cargo.toml index cc2e7ab..09cb494 100644 --- a/Cargo.toml +++ b/Cargo.toml
@@ -9,8 +9,7 @@ "bench/report", "cli", "examples", - # todo - #"integration", + "integration", "sdk", "server", "tools",
diff --git a/integration/tests/examples/mod.rs b/integration/tests/examples/mod.rs index 3f9e8dd..7d2aa01 100644 --- a/integration/tests/examples/mod.rs +++ b/integration/tests/examples/mod.rs
@@ -174,7 +174,7 @@ 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 dfba2f1..4682187 100644 --- a/integration/tests/mod.rs +++ b/integration/tests/mod.rs
@@ -33,7 +33,7 @@ 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 3136c10..19878f4 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 @@ 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 @@ .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 bfbb570..c0d9f90 100644 --- a/integration/tests/server/scenarios/message_size_scenario.rs +++ b/integration/tests/server/scenarios/message_size_scenario.rs
@@ -10,7 +10,7 @@ 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 @@ 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 @@ "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 d37a81d..eafe2c4 100644 --- a/integration/tests/server/scenarios/mod.rs +++ b/integration/tests/server/scenarios/mod.rs
@@ -28,7 +28,7 @@ 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 266544b..27a62a6 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 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 f6d0f74..e027d0e 100644 --- a/integration/tests/server/scenarios/system_scenario.rs +++ b/integration/tests/server/scenarios/system_scenario.rs
@@ -273,7 +273,7 @@ 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 91e64cf..4254972 100644 --- a/integration/tests/streaming/segment.rs +++ b/integration/tests/streaming/segment.rs
@@ -317,9 +317,11 @@ 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 84c2174..72bd11c 100644 --- a/sdk/src/error.rs +++ b/sdk/src/error.rs
@@ -312,8 +312,8 @@ 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 09c543b..ae88365 100644 --- a/sdk/src/messages/mod.rs +++ b/sdk/src/messages/mod.rs
@@ -7,7 +7,7 @@ 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 ca848a6..4afb35a 100644 --- a/sdk/src/messages/polled_messages.rs +++ b/sdk/src/messages/polled_messages.rs
@@ -72,7 +72,6 @@ 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 0fc99d6..e329555 100644 --- a/sdk/src/messages/send_messages.rs +++ b/sdk/src/messages/send_messages.rs
@@ -125,8 +125,7 @@ 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 500b1d2..0ac9825 100644 --- a/sdk/src/models/messaging/indexes.rs +++ b/sdk/src/models/messaging/indexes.rs
@@ -38,6 +38,11 @@ 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 5289206..3ff535b 100644 --- a/sdk/src/models/messaging/messages_batch.rs +++ b/sdk/src/models/messaging/messages_batch.rs
@@ -413,7 +413,7 @@ } 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 @@ 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 15b30a8..743a941 100644 --- a/sdk/src/prelude.rs +++ b/sdk/src/prelude.rs
@@ -31,6 +31,7 @@ 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 9574258..1214c69 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 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 @@ // 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 @@ 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 5368ca1..6a7a933 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 @@ 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 a52b137..b0e44d7 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 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 710e4a4..bcb3289 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 @@ 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 aee3253..c3c4d7f 100644 --- a/server/src/quic/quic_sender.rs +++ b/server/src/quic/quic_sender.rs
@@ -95,7 +95,11 @@ 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 9432561..c5f1423 100644 --- a/server/src/streaming/partitions/messages.rs +++ b/server/src/streaming/partitions/messages.rs
@@ -116,7 +116,7 @@ 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 a4e0eb4..d3094a9 100644 --- a/server/src/streaming/segments/indexes/index_writer.rs +++ b/server/src/streaming/segments/indexes/index_writer.rs
@@ -87,7 +87,7 @@ 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 91bf98b..7b1dd6d 100644 --- a/server/src/streaming/segments/indexes/indexes_mut.rs +++ b/server/src/streaming/segments/indexes/indexes_mut.rs
@@ -28,6 +28,18 @@ } } + /// 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 @@ } /// 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 @@ 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 @@ 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 33e9150..397ef73 100644 --- a/server/src/streaming/segments/messages_accumulator.rs +++ b/server/src/streaming/segments/messages_accumulator.rs
@@ -108,6 +108,8 @@ /// /// 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 8cdc94c..b23b832 100644 --- a/server/src/streaming/segments/reading_messages.rs +++ b/server/src/streaming/segments/reading_messages.rs
@@ -9,12 +9,18 @@ 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 @@ 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 @@ 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 2567271..319cbe6 100644 --- a/server/src/streaming/segments/segment.rs +++ b/server/src/streaming/segments/segment.rs
@@ -151,7 +151,7 @@ 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 @@ 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 7bb16e2..e2c3a2a 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 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 @@ 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 ed76810..8599ee7 100644 --- a/server/src/streaming/segments/writing_messages.rs +++ b/server/src/streaming/segments/writing_messages.rs
@@ -129,6 +129,8 @@ .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 41db1ae..f86c9b9 100644 --- a/server/src/tcp/sender.rs +++ b/server/src/tcp/sender.rs
@@ -65,7 +65,11 @@ 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 @@ 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(()) }