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 2025/07/14 09:07:39 UTC
(iggy) branch fix_segment_integration updated (3f06e543 -> 522e4e50)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch fix_segment_integration
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 3f06e543 temp
add 3e371b6d feat(io_uring): fix busy poll (#2002)
new 522e4e50 feat(io_uring): fix segment integration tests
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (3f06e543)
\
N -- N -- N refs/heads/fix_segment_integration (522e4e50)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../tests/streaming/common/test_setup.rs | 5 ++-
core/integration/tests/streaming/segment.rs | 25 +++++++++-----
.../handlers/streams/purge_stream_handler.rs | 6 ++--
.../binary/handlers/topics/purge_topic_handler.rs | 4 +--
core/server/src/bootstrap.rs | 5 ++-
core/server/src/shard/mod.rs | 39 ++++++++++++++--------
core/server/src/shard/system/segments.rs | 3 +-
core/server/src/shard/system/streams.rs | 5 +--
core/server/src/shard/system/topics.rs | 11 +++---
.../src/streaming/segments/indexes/index_reader.rs | 13 ++++----
.../streaming/segments/messages/messages_reader.rs | 6 ++--
.../src/streaming/segments/reading_messages.rs | 2 +-
core/server/src/streaming/segments/segment.rs | 2 ++
13 files changed, 78 insertions(+), 48 deletions(-)
(iggy) 01/01: feat(io_uring): fix segment integration tests
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch fix_segment_integration
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 522e4e507929efd5aceef3074037277c39b6dc60
Author: numminex <g....@gmail.com>
AuthorDate: Mon Jul 14 11:05:02 2025 +0200
feat(io_uring): fix segment integration tests
---
Cargo.lock | 1 +
Cargo.toml | 1 +
core/integration/Cargo.toml | 1 +
.../tests/streaming/common/test_setup.rs | 5 +-
core/integration/tests/streaming/segment.rs | 56 +++++++++--------
core/server/Cargo.toml | 2 +-
.../handlers/streams/purge_stream_handler.rs | 6 +-
.../binary/handlers/topics/purge_topic_handler.rs | 4 +-
core/server/src/shard/mod.rs | 70 ++++++++++++++++++++--
core/server/src/shard/system/segments.rs | 4 +-
core/server/src/shard/system/streams.rs | 5 +-
core/server/src/shard/system/topics.rs | 11 ++--
.../src/streaming/segments/indexes/index_reader.rs | 13 ++--
.../streaming/segments/messages/messages_reader.rs | 6 +-
.../src/streaming/segments/reading_messages.rs | 2 +-
core/server/src/streaming/segments/segment.rs | 2 +
16 files changed, 133 insertions(+), 56 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 6c09fa09..06abcf63 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4275,6 +4275,7 @@ dependencies = [
"async-trait",
"bytes",
"chrono",
+ "compio",
"ctor",
"derive_more 2.0.1",
"env_logger",
diff --git a/Cargo.toml b/Cargo.toml
index 6a2c27f4..881952a2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -117,6 +117,7 @@ sysinfo = "0.35.2"
tempfile = "3.20.0"
thiserror = "2.0.12"
tokio = { version = "1.45.1", features = ["full"] }
+compio = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be", features = ["runtime", "macros", "io-uring", "time", "rustls"] }
tokio-rustls = "0.26.2"
toml = "0.8.23"
tracing = "0.1.41"
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index b17d3d0b..7c241429 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -51,6 +51,7 @@ server = { workspace = true }
tempfile = { workspace = true }
test-case = { workspace = true }
tokio = { workspace = true }
+compio = { workspace = true }
twox-hash = { workspace = true }
uuid = { workspace = true }
zip = { workspace = true }
diff --git a/core/integration/tests/streaming/common/test_setup.rs b/core/integration/tests/streaming/common/test_setup.rs
index 19363249..88c32194 100644
--- a/core/integration/tests/streaming/common/test_setup.rs
+++ b/core/integration/tests/streaming/common/test_setup.rs
@@ -16,13 +16,15 @@
* under the License.
*/
+use compio::fs;
+use log::error;
+use server::bootstrap::create_directories;
use server::configs::system::SystemConfig;
use server::streaming::persistence::persister::{FileWithSyncPersister, PersisterKind};
use server::streaming::storage::SystemStorage;
use server::streaming::utils::MemoryPool;
use std::rc::Rc;
use std::sync::Arc;
-use tokio::fs;
use uuid::Uuid;
pub struct TestSetup {
@@ -42,6 +44,7 @@ impl TestSetup {
let config = Arc::new(config);
fs::create_dir(config.get_system_path()).await.unwrap();
+ create_directories(&config).await.unwrap();
let persister = PersisterKind::FileWithSync(FileWithSyncPersister {});
let storage = Rc::new(SystemStorage::new(config.clone(), Arc::new(persister)));
MemoryPool::init_pool(config.clone());
diff --git a/core/integration/tests/streaming/segment.rs b/core/integration/tests/streaming/segment.rs
index 04155cb8..cc01eb71 100644
--- a/core/integration/tests/streaming/segment.rs
+++ b/core/integration/tests/streaming/segment.rs
@@ -19,22 +19,25 @@
use crate::streaming::common::test_setup::TestSetup;
use crate::streaming::create_message;
use bytes::Bytes;
+use compio::fs;
+use compio::time::sleep;
use iggy::prelude::locking::IggyRwLockFn;
use iggy::prelude::*;
-use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
+use log::error;
+use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig, ServerConfig};
use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
+use server::shard::IggyShard;
use server::streaming::segments::*;
use server::streaming::session::Session;
use std::fs::DirEntry;
use std::net::{Ipv4Addr, SocketAddr};
+use std::rc::Rc;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::Duration;
-use tokio::fs;
-use tokio::time::sleep;
-#[tokio::test]
+#[compio::test]
async fn should_persist_segment() {
let setup = TestSetup::init().await;
let stream_id = 1;
@@ -72,7 +75,7 @@ async fn should_persist_segment() {
}
}
-#[tokio::test]
+#[compio::test]
async fn should_load_existing_segment_from_disk() {
let setup = TestSetup::init().await;
let stream_id = 1;
@@ -142,7 +145,7 @@ async fn should_load_existing_segment_from_disk() {
}
}
-#[tokio::test]
+#[compio::test]
async fn should_persist_and_load_segment_with_messages() {
let setup = TestSetup::init().await;
let stream_id = 1;
@@ -215,7 +218,7 @@ async fn should_persist_and_load_segment_with_messages() {
assert_eq!(messages.count(), messages_count);
}
-#[tokio::test]
+#[compio::test]
async fn given_at_least_one_not_expired_message_segment_should_not_be_expired() {
let config = SystemConfig {
partition: PartitionConfig {
@@ -307,9 +310,7 @@ async fn given_at_least_one_not_expired_message_segment_should_not_be_expired()
);
}
-/*
-//TODO: Fixme use shard instead of system
-#[tokio::test]
+#[compio::test]
async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Error>> {
let config = SystemConfig {
segment: SegmentConfig {
@@ -319,11 +320,12 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
..Default::default()
};
let setup = TestSetup::init_with_config(config).await;
- let mut system = System::new(
- setup.config.clone(),
- DataMaintenanceConfig::default(),
- PersonalAccessTokenConfig::default(),
- );
+ let server_config = ServerConfig {
+ system: setup.config.clone(),
+ ..Default::default()
+ };
+ let shard = IggyShard::default_from_config(server_config);
+ shard.init().await.unwrap();
// Properties
let stream_id = Identifier::numeric(1)?;
@@ -341,14 +343,16 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
.await;
let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- system.init().await.unwrap();
+ let session = Rc::new(session);
+ shard.init().await.unwrap();
+ shard.add_active_session(session.clone());
- system
+ shard
.create_stream(&session, Some(stream_id.get_u32_value()?), stream_name)
.await
.unwrap();
- system
+ let result = shard
.create_topic(
&session,
&stream_id,
@@ -360,9 +364,13 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
MaxTopicSize::default(),
None,
)
- .await?;
+ .await;
+ if let Some(error) = result.err() {
+ error!("Failed to create topic: {}", error);
+ }
- let topic = system.find_topic(&session, &stream_id, &topic_id)?;
+ let stream = shard.find_stream(&session, &stream_id)?;
+ let topic = shard.find_topic(&session, &stream, &topic_id)?;
let partitions = topic.get_partitions();
let partition = partitions
.first()
@@ -384,12 +392,13 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition_lock.append_messages(batch, None).await.unwrap();
+ partition_lock.append_messages(batch).await.unwrap();
drop(partition_lock);
sleep(Duration::from_millis(100)).await;
}
+ drop(stream);
let partition_path = setup.config.get_partition_path(
stream_id.get_u32_value()?,
@@ -418,7 +427,7 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
println!("Initial segment offsets (sorted): {initial_offsets:?}");
let first_keep_count = 3usize;
- system
+ shard
.delete_segments(
&session,
&stream_id,
@@ -449,7 +458,7 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
// Second attempt to delete segments - keeping only 1 closed segment + 1 open
let second_keep_count = 1usize;
- system
+ shard
.delete_segments(
&session,
&stream_id,
@@ -497,7 +506,6 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
Ok(())
}
-*/
// Helper function to extract segment offsets from DirEntry list
fn get_segment_offsets(segments: &[DirEntry]) -> Vec<u64> {
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 47738185..b171652d 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -67,7 +67,6 @@ lending-iterator = "0.1.7"
hash32 = "1.0.0"
mimalloc = { workspace = true, optional = true }
moka = { version = "0.12.10", features = ["future"] }
-compio = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be", features = ["runtime", "macros", "io-uring", "time", "rustls"] }
nix = { version = "0.30", features = ["fs"] }
once_cell = "1.21.3"
opentelemetry = { version = "0.30.0", features = ["trace", "logs"] }
@@ -107,6 +106,7 @@ sysinfo = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
+compio = { workspace = true }
tokio-rustls = { workspace = true }
tokio-util = { workspace = true }
toml = { workspace = true }
diff --git a/core/server/src/binary/handlers/streams/purge_stream_handler.rs b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
index 6f8f06e8..1bb5fb39 100644
--- a/core/server/src/binary/handlers/streams/purge_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
@@ -19,8 +19,8 @@
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::{handlers::streams::COMPONENT, sender::SenderKind};
-use crate::shard::transmission::event::ShardEvent;
use crate::shard::IggyShard;
+use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -53,7 +53,9 @@ impl ServerCommandHandler for PurgeStream {
format!("{COMPONENT} (error: {error}) - failed to purge stream with id: {stream_id}, session: {session}")
})?;
- let event = ShardEvent::PurgedStream { stream_id: self.stream_id.clone() };
+ let event = ShardEvent::PurgedStream {
+ stream_id: self.stream_id.clone(),
+ };
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
shard
diff --git a/core/server/src/binary/handlers/topics/purge_topic_handler.rs b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
index 160fac9a..101c7973 100644
--- a/core/server/src/binary/handlers/topics/purge_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
@@ -45,7 +45,7 @@ impl ServerCommandHandler for PurgeTopic {
debug!("session: {session}, command: {self}");
let topic_id = self.topic_id.clone();
let stream_id = self.stream_id.clone();
-
+
shard
.purge_topic(session, &self.stream_id, &self.topic_id)
.await
@@ -56,7 +56,7 @@ impl ServerCommandHandler for PurgeTopic {
)
})?;
- let event = crate::shard::transmission::event::ShardEvent::PurgedTopic {
+ let event = crate::shard::transmission::event::ShardEvent::PurgedTopic {
stream_id: self.stream_id.clone(),
topic_id: self.topic_id.clone(),
};
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 4d0f089e..b8ba22b7 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -28,10 +28,15 @@ use ahash::{AHashMap, AHashSet, HashMap};
use builder::IggyShardBuilder;
use error_set::ErrContext;
use futures::future::try_join_all;
-use iggy_common::{EncryptorKind, Identifier, IggyError, UserId, locking::IggyRwLockFn};
+use iggy_common::{
+ EncryptorKind, Identifier, IggyError, Permissions, UserId, UserStatus,
+ defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME},
+ locking::IggyRwLockFn,
+};
use namespace::IggyNamespace;
use std::{
cell::{Cell, RefCell},
+ future::Future,
pin::Pin,
rc::Rc,
str::FromStr,
@@ -156,13 +161,67 @@ impl IggyShard {
Default::default()
}
+ pub fn default_from_config(server_config: ServerConfig) -> Self {
+ use crate::bootstrap::resolve_persister;
+ use crate::state::file::FileState;
+ use crate::streaming::diagnostics::metrics::Metrics;
+ use crate::streaming::storage::SystemStorage;
+ use crate::versioning::SemanticVersion;
+
+ let version = SemanticVersion::current().expect("Invalid version");
+ let persister = resolve_persister(server_config.system.partition.enforce_fsync);
+ let storage = Rc::new(SystemStorage::new(
+ server_config.system.clone(),
+ persister.clone(),
+ ));
+
+ let state_path = server_config.system.get_state_messages_file_path();
+ let file_state = FileState::new(&state_path, &version, persister, None);
+ let state = crate::state::StateKind::File(file_state);
+
+ let (stop_sender, stop_receiver) = async_channel::unbounded();
+
+ let shard = Self {
+ id: 0, // Default shard ID
+ shards: Vec::new(), // No other shards in default config
+ shards_table: Default::default(),
+ version,
+ streams: Default::default(),
+ streams_ids: Default::default(),
+ storage,
+ state,
+ encryptor: None,
+ config: server_config,
+ client_manager: Default::default(),
+ active_sessions: Default::default(),
+ permissioner: Default::default(),
+ users: Default::default(),
+ metrics: Metrics::init(),
+ messages_receiver: Cell::new(None),
+ stop_receiver,
+ stop_sender,
+ task_registry: TaskRegistry::new(),
+ is_shutting_down: AtomicBool::new(false),
+ };
+ let user = User::root(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD);
+ shard
+ .create_user_bypass_auth(
+ &user.username,
+ &user.password,
+ UserStatus::Active,
+ Some(Permissions::root()),
+ )
+ .unwrap();
+ shard
+ }
+
pub async fn init(&self) -> Result<(), IggyError> {
let now = Instant::now();
self.load_version().await?;
let SystemState { users, streams } = self.load_state().await?;
- self.load_users(users.into_values().collect()).await;
- self.load_streams(streams.into_values().collect()).await;
+ let _ = self.load_users(users.into_values().collect()).await;
+ let _ = self.load_streams(streams.into_values().collect()).await;
//TODO: Fix the archiver.
/*
@@ -534,7 +593,6 @@ impl IggyShard {
*max_topic_size,
*replication_factor,
)
- .await
}
ShardEvent::LoginUser {
client_id,
@@ -689,7 +747,7 @@ impl IggyShard {
.await?;
Ok(())
}
- ShardEvent::PurgedStream { stream_id,} => {
+ ShardEvent::PurgedStream { stream_id } => {
self.purge_stream_bypass_auth(stream_id).await?;
Ok(())
}
@@ -699,7 +757,7 @@ impl IggyShard {
} => {
self.purge_topic_bypass_auth(stream_id, topic_id).await?;
Ok(())
- },
+ }
ShardEvent::DeletedTopic {
stream_id,
topic_id,
diff --git a/core/server/src/shard/system/segments.rs b/core/server/src/shard/system/segments.rs
index 5b907f05..0c7b853b 100644
--- a/core/server/src/shard/system/segments.rs
+++ b/core/server/src/shard/system/segments.rs
@@ -87,6 +87,7 @@ impl IggyShard {
.map(|segment| (segment.start_offset(), segment.get_messages_count()))
.collect::<Vec<_>>();
+
// Delete the segments in sequence.
let (deleted_segments_count, deleted_messages_count) = {
let mut segments_count = 0;
@@ -103,9 +104,8 @@ impl IggyShard {
(segments_count, messages_count)
};
- drop(partition);
- drop(topic);
drop(stream);
+ drop(partition);
let mut stream = self
.get_stream_mut(stream_id)
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index f5e823c1..b4b1b63a 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -414,10 +414,7 @@ impl IggyShard {
Ok(())
}
- pub async fn purge_stream_bypass_auth(
- &self,
- stream_id: &Identifier,
- ) -> Result<(), IggyError> {
+ pub async fn purge_stream_bypass_auth(&self, stream_id: &Identifier) -> Result<(), IggyError> {
let stream = self.get_stream(stream_id).with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
})?;
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index d71add8e..bfe974cf 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -102,7 +102,7 @@ impl IggyShard {
Ok(Some(topic))
}
- pub async fn create_topic_bypass_auth(
+ pub fn create_topic_bypass_auth(
&self,
stream_id: &Identifier,
topic_id: Option<u32>,
@@ -445,7 +445,7 @@ impl IggyShard {
session.get_user_id(),
)
})?;
-
+
self.purge_topic_base(topic.stream_id, topic.topic_id).await
}
@@ -462,14 +462,15 @@ impl IggyShard {
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}")
})?;
-
- self.purge_topic_base(stream.stream_id, topic.topic_id).await
+
+ self.purge_topic_base(stream.stream_id, topic.topic_id)
+ .await
}
async fn purge_topic_base(&self, stream_id: u32, topic_id: u32) -> Result<(), IggyError> {
let stream = self.get_stream(&Identifier::numeric(stream_id)?)?;
let topic = stream.get_topic(&Identifier::numeric(topic_id)?)?;
-
+
for partition in topic.get_partitions() {
let mut partition = partition.write().await;
let partition_id = partition.partition_id;
diff --git a/core/server/src/streaming/segments/indexes/index_reader.rs b/core/server/src/streaming/segments/indexes/index_reader.rs
index ce235ef4..3275f567 100644
--- a/core/server/src/streaming/segments/indexes/index_reader.rs
+++ b/core/server/src/streaming/segments/indexes/index_reader.rs
@@ -20,12 +20,11 @@ use super::IggyIndexesMut;
use crate::streaming::utils::PooledBuffer;
use bytes::BytesMut;
use compio::{
- BufResult,
- fs::{File, OpenOptions},
- io::AsyncReadAtExt,
+ buf::{IntoInner, IoBuf}, fs::{File, OpenOptions}, io::AsyncReadAtExt, BufResult
};
use error_set::ErrContext;
use iggy_common::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView};
+use ring::error;
use std::{
io::ErrorKind,
os::unix::fs::FileExt,
@@ -150,6 +149,7 @@ impl IndexReader {
{
Ok(buf) => buf,
Err(error) if error.kind() == ErrorKind::UnexpectedEof => {
+ error!("Unexpected EOF while reading indexes");
return Ok(None);
}
Err(error) => {
@@ -337,9 +337,10 @@ impl IndexReader {
use_pool: bool,
) -> Result<PooledBuffer, std::io::Error> {
if use_pool {
- let mut buf = PooledBuffer::with_capacity(len as usize);
- unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
+ let len = len as usize;
+ let buf = PooledBuffer::with_capacity(len as usize);
+ let (result, buf) = self.file.read_exact_at(buf.slice(..len), offset as u64).await.into();
+ let buf = buf.into_inner();
result?;
Ok(buf)
} else {
diff --git a/core/server/src/streaming/segments/messages/messages_reader.rs b/core/server/src/streaming/segments/messages/messages_reader.rs
index e81eb793..fea3ca37 100644
--- a/core/server/src/streaming/segments/messages/messages_reader.rs
+++ b/core/server/src/streaming/segments/messages/messages_reader.rs
@@ -19,6 +19,7 @@
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
use crate::streaming::utils::PooledBuffer;
use bytes::BytesMut;
+use compio::buf::{IntoInner, IoBuf};
use compio::fs::{File, OpenOptions};
use compio::io::AsyncReadAtExt;
use error_set::ErrContext;
@@ -178,8 +179,9 @@ impl MessagesReader {
) -> Result<PooledBuffer, std::io::Error> {
if use_pool {
let mut buf = PooledBuffer::with_capacity(len as usize);
- unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
+ let len = len as usize;
+ let (result, buf) = self.file.read_exact_at(buf.slice(..len), offset as u64).await.into();
+ let buf = buf.into_inner();
result?;
Ok(buf)
} else {
diff --git a/core/server/src/streaming/segments/reading_messages.rs b/core/server/src/streaming/segments/reading_messages.rs
index fa46c183..d72b1591 100644
--- a/core/server/src/streaming/segments/reading_messages.rs
+++ b/core/server/src/streaming/segments/reading_messages.rs
@@ -21,7 +21,7 @@ use crate::streaming::segments::segment::Segment;
use error_set::ErrContext;
use iggy_common::{IggyByteSize, IggyError};
use std::sync::atomic::Ordering;
-use tracing::trace;
+use tracing::{error, trace};
const COMPONENT: &str = "STREAMING_SEGMENT";
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index dfe7bc21..9f7dcea5 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -28,6 +28,7 @@ use iggy_common::IggyByteSize;
use iggy_common::IggyError;
use iggy_common::IggyExpiry;
use iggy_common::IggyTimestamp;
+use tracing::error;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
@@ -285,6 +286,7 @@ impl Segment {
let last_messages = last_messages.unwrap();
if last_messages.is_empty() {
+ error!("empty messages");
return false;
}