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 15:13:33 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): fix shard integration tests (#2008)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new fed96239 feat(io_uring): fix shard integration tests (#2008)
fed96239 is described below
commit fed96239e08e1cc9ea5f9d829a579c0e530acc09
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Mon Jul 14 17:13:30 2025 +0200
feat(io_uring): fix shard integration tests (#2008)
---
core/integration/tests/streaming/shard.rs | 150 +++++++++++++--------
core/sdk/src/clients/client.rs | 1 -
core/server/src/io/fs_utils.rs | 22 ++-
core/server/src/shard/mod.rs | 3 +-
core/server/src/shard/system/segments.rs | 1 -
core/server/src/shard/system/streams.rs | 34 +++--
.../src/streaming/segments/indexes/index_reader.rs | 11 +-
.../streaming/segments/messages/messages_reader.rs | 6 +-
core/server/src/streaming/segments/segment.rs | 2 +-
9 files changed, 149 insertions(+), 81 deletions(-)
diff --git a/core/integration/tests/streaming/shard.rs b/core/integration/tests/streaming/shard.rs
index 454a5841..ec2e26da 100644
--- a/core/integration/tests/streaming/shard.rs
+++ b/core/integration/tests/streaming/shard.rs
@@ -17,107 +17,147 @@
*/
use crate::streaming::common::test_setup::TestSetup;
+use compio::fs;
use iggy::prelude::Identifier;
-use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
+use iggy_common::locking::IggyRwLockFn;
+use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig, ServerConfig};
+use server::configs::system::SystemConfig;
+use server::shard::IggyShard;
use server::streaming::session::Session;
use std::net::{Ipv4Addr, SocketAddr};
-use tokio::fs;
+use std::rc::Rc;
//TODO: Fix me use shard instead of system
-/*
-#[tokio::test]
+#[compio::test]
async fn should_initialize_system_and_base_directories() {
let setup = TestSetup::init().await;
- let mut system = System::new(
- setup.config.clone(),
- DataMaintenanceConfig::default(),
- PersonalAccessTokenConfig::default(),
- );
-
- system.init().await.unwrap();
-
- let mut dir_entries = fs::read_dir(&setup.config.path).await.unwrap();
+ let server_config = ServerConfig {
+ system: setup.config.clone(),
+ ..Default::default()
+ };
+ let shard = IggyShard::default_from_config(server_config);
+ shard.init().await.unwrap();
+
+ let mut dir_entries = server::io::fs_utils::walk_dir(&setup.config.path)
+ .await
+ .unwrap();
let mut names = Vec::new();
- while let Some(entry) = dir_entries.next_entry().await.unwrap() {
- let metadata = entry.metadata().await.unwrap();
- assert!(metadata.is_dir());
- names.push(entry.file_name().into_string().unwrap());
+ for dir_entry in dir_entries.iter_mut() {
+ if dir_entry.is_dir {
+ names.push(dir_entry.name.clone().unwrap());
+ } else {
+ dir_entry.name = Some(
+ dir_entry
+ .path
+ .file_name()
+ .unwrap()
+ .to_string_lossy()
+ .to_string(),
+ );
+ }
}
- assert_eq!(names.len(), 3);
- assert!(names.contains(&setup.config.stream.path));
+ // Root + streams + state + runtime
+ assert_eq!(names.len(), 4);
+ let mut has_stream = false;
+ for name in names {
+ if name.contains(&setup.config.stream.path) {
+ has_stream = true;
+ }
+ }
+ assert!(has_stream);
}
-#[tokio::test]
+#[compio::test]
async fn should_create_and_persist_stream() {
let setup = TestSetup::init().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);
let stream_id = 1;
let stream_name = "test";
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
- .create_stream(&session, Some(stream_id), stream_name)
- .await
+ shard
+ .create_stream_bypass_auth(Some(stream_id), stream_name)
+ .unwrap();
+ let stream = shard
+ .get_stream(&Identifier::numeric(stream_id).unwrap())
.unwrap();
+ stream.persist().await.unwrap();
assert_persisted_stream(&setup.config.get_streams_path(), stream_id).await;
}
-#[tokio::test]
+#[compio::test]
async fn should_create_and_persist_stream_with_automatically_generated_id() {
let setup = TestSetup::init().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);
let stream_id = 1;
let stream_name = "test";
+ shard.init().await.unwrap();
let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- system.init().await.unwrap();
+ let session = Rc::new(session);
+ shard.add_active_session(session.clone());
- system
- .create_stream(&session, None, stream_name)
- .await
+ let stream_id = shard.create_stream_bypass_auth(None, stream_name).unwrap();
+ let stream = shard
+ .get_stream(&Identifier::numeric(stream_id).unwrap())
.unwrap();
+ stream.persist().await.unwrap();
assert_persisted_stream(&setup.config.get_streams_path(), stream_id).await;
}
-#[tokio::test]
+#[compio::test]
async fn should_delete_persisted_stream() {
let setup = TestSetup::init().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);
let stream_id = 1;
let stream_name = "test";
let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- system.init().await.unwrap();
- system
- .create_stream(&session, Some(stream_id), stream_name)
- .await
+ let session = Rc::new(session);
+ shard.init().await.unwrap();
+ shard.add_active_session(session.clone());
+
+ let id = shard
+ .create_stream_bypass_auth(Some(stream_id), stream_name)
.unwrap();
+ let stream = shard.get_stream(&Identifier::numeric(id).unwrap()).unwrap();
+ stream.persist().await.unwrap();
+ drop(stream);
assert_persisted_stream(&setup.config.get_streams_path(), stream_id).await;
- let stream_path = system
+ let stream_path = shard
.get_stream(&Identifier::numeric(stream_id).unwrap())
.unwrap()
.path
.clone();
- system
- .delete_stream(&session, &Identifier::numeric(1).unwrap())
- .await
+ let stream = shard
+ .delete_stream_bypass_auth(&Identifier::numeric(stream_id).unwrap())
.unwrap();
-
+ for topic in stream.get_topics() {
+ let partitions = topic.get_partitions();
+ for partition in partitions {
+ let mut part = partition.write().await;
+ part.delete().await.unwrap();
+ }
+ topic.delete().await.unwrap();
+ }
+ stream.delete().await.unwrap();
assert!(fs::metadata(stream_path).await.is_err());
}
@@ -128,5 +168,3 @@ async fn assert_persisted_stream(streams_path: &str, stream_id: u32) {
let stream_metadata = fs::metadata(stream_path).await.unwrap();
assert!(stream_metadata.is_dir());
}
-
-*/
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 0482a2b8..6189a8da 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -70,7 +70,6 @@ impl IggyClient {
IggyClientBuilder::from_connection_string(connection_string)
}
-
/// Creates a new `IggyClient` with the provided client implementation for the specific transport.
pub fn new(client: ClientWrapper) -> Self {
let client = IggyRwLock::new(client);
diff --git a/core/server/src/io/fs_utils.rs b/core/server/src/io/fs_utils.rs
index 6bbc1b7f..fbb95a60 100644
--- a/core/server/src/io/fs_utils.rs
+++ b/core/server/src/io/fs_utils.rs
@@ -24,18 +24,24 @@ use std::path::{Path, PathBuf};
pub struct DirEntry {
pub path: PathBuf,
pub is_dir: bool,
+ pub name: Option<String>,
}
impl DirEntry {
- fn file(path: PathBuf) -> Self {
+ fn file(path: PathBuf, name: Option<String>) -> Self {
Self {
path,
is_dir: false,
+ name,
}
}
- fn dir(path: PathBuf) -> Self {
- Self { path, is_dir: true }
+ fn dir(path: PathBuf, name: Option<String>) -> Self {
+ Self {
+ path,
+ is_dir: true,
+ name,
+ }
}
}
@@ -59,7 +65,10 @@ pub async fn walk_dir(root: impl AsRef<Path>) -> io::Result<Vec<DirEntry>> {
let mut stack = vec![root.to_path_buf()];
while let Some(current_dir) = stack.pop() {
- directories.push(DirEntry::dir(current_dir.clone()));
+ directories.push(DirEntry::dir(
+ current_dir.clone(),
+ current_dir.to_str().map(|s| s.to_string()),
+ ));
for entry in std::fs::read_dir(¤t_dir)? {
let entry = entry?;
@@ -69,7 +78,10 @@ pub async fn walk_dir(root: impl AsRef<Path>) -> io::Result<Vec<DirEntry>> {
if metadata.is_dir() {
stack.push(entry_path);
} else {
- files.push(DirEntry::file(entry_path));
+ files.push(DirEntry::file(
+ entry_path,
+ entry.file_name().into_string().ok(),
+ ));
}
}
}
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index b8ba22b7..b884a92d 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -570,7 +570,8 @@ impl IggyShard {
async fn handle_event(&self, event: Arc<ShardEvent>) -> Result<(), IggyError> {
match &*event {
ShardEvent::CreatedStream { stream_id, name } => {
- self.create_stream_bypass_auth(*stream_id, name)
+ self.create_stream_bypass_auth(*stream_id, name)?;
+ Ok(())
}
ShardEvent::CreatedTopic {
stream_id,
diff --git a/core/server/src/shard/system/segments.rs b/core/server/src/shard/system/segments.rs
index 0c7b853b..e895f6fd 100644
--- a/core/server/src/shard/system/segments.rs
+++ b/core/server/src/shard/system/segments.rs
@@ -87,7 +87,6 @@ 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;
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index b4b1b63a..cba0c14a 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -193,13 +193,13 @@ impl IggyShard {
&self,
stream_id: Option<u32>,
name: &str,
- ) -> Result<(), IggyError> {
+ ) -> Result<u32, IggyError> {
let stream = self.create_stream_base(stream_id, name)?;
let id = stream.stream_id;
self.streams_ids.borrow_mut().insert(name.to_owned(), id);
self.streams.borrow_mut().insert(id, stream);
self.metrics.increment_streams(1);
- Ok(())
+ Ok(id)
}
pub async fn create_stream(
@@ -341,22 +341,30 @@ impl IggyShard {
}
pub fn delete_stream_bypass_auth(&self, id: &Identifier) -> Result<Stream, IggyError> {
- let stream = self.get_stream(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
- })?;
- let stream_id = stream.stream_id;
- let stream_name = stream.name.clone();
+ let stream_id;
+ let stream_name;
+ {
+ let stream = self.get_stream(id).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
+ })?;
+ stream_id = stream.stream_id;
+ stream_name = stream.name.clone();
+ }
let stream = self.delete_stream_base(stream_id, stream_name)?;
Ok(stream)
}
pub fn delete_stream(&self, session: &Session, id: &Identifier) -> Result<Stream, IggyError> {
self.ensure_authenticated(session)?;
- let stream = self.get_stream(id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
- })?;
- let stream_id = stream.stream_id;
- let stream_name = stream.name.clone();
+ let stream_id;
+ let stream_name;
+ {
+ let stream = self.get_stream(id).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {id}")
+ })?;
+ stream_id = stream.stream_id;
+ stream_name = stream.name.clone();
+ }
self.permissioner
.borrow()
.delete_stream(session.get_user_id(), stream_id)
@@ -364,7 +372,7 @@ impl IggyShard {
format!(
"{COMPONENT} (error: {error}) - permission denied to delete stream for user {}, stream ID: {}",
session.get_user_id(),
- stream.stream_id,
+ stream_id,
)
})?;
let stream = self.delete_stream_base(stream_id, stream_name)?;
diff --git a/core/server/src/streaming/segments/indexes/index_reader.rs b/core/server/src/streaming/segments/indexes/index_reader.rs
index 3275f567..48ed12b3 100644
--- a/core/server/src/streaming/segments/indexes/index_reader.rs
+++ b/core/server/src/streaming/segments/indexes/index_reader.rs
@@ -20,7 +20,10 @@ use super::IggyIndexesMut;
use crate::streaming::utils::PooledBuffer;
use bytes::BytesMut;
use compio::{
- buf::{IntoInner, IoBuf}, fs::{File, OpenOptions}, io::AsyncReadAtExt, BufResult
+ BufResult,
+ buf::{IntoInner, IoBuf},
+ fs::{File, OpenOptions},
+ io::AsyncReadAtExt,
};
use error_set::ErrContext;
use iggy_common::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView};
@@ -339,7 +342,11 @@ impl IndexReader {
if use_pool {
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 (result, buf) = self
+ .file
+ .read_exact_at(buf.slice(..len), offset as u64)
+ .await
+ .into();
let buf = buf.into_inner();
result?;
Ok(buf)
diff --git a/core/server/src/streaming/segments/messages/messages_reader.rs b/core/server/src/streaming/segments/messages/messages_reader.rs
index fea3ca37..bd318fc9 100644
--- a/core/server/src/streaming/segments/messages/messages_reader.rs
+++ b/core/server/src/streaming/segments/messages/messages_reader.rs
@@ -180,7 +180,11 @@ impl MessagesReader {
if use_pool {
let mut buf = PooledBuffer::with_capacity(len as usize);
let len = len as usize;
- let (result, buf) = self.file.read_exact_at(buf.slice(..len), offset as u64).await.into();
+ let (result, buf) = self
+ .file
+ .read_exact_at(buf.slice(..len), offset as u64)
+ .await
+ .into();
let buf = buf.into_inner();
result?;
Ok(buf)
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index fdca6be6..a3ac8588 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -28,10 +28,10 @@ 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};
+use tracing::error;
use tracing::{info, warn};
const SIZE_16MB: usize = 16 * 1024 * 1024;