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/10/16 08:17:59 UTC
(iggy) branch io_uring_tpc_broadcast created (now 9ab29580d)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch io_uring_tpc_broadcast
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 9ab29580d fixes
This branch includes the following new commits:
new 9ab29580d fixes
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.
(iggy) 01/01: fixes
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch io_uring_tpc_broadcast
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9ab29580d888c3d5ae129a962a4ec3260d7c6a54
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 16 10:17:44 2025 +0200
fixes
---
core/integration/tests/server/general.rs | 18 ++-
.../tests/server/scenarios/concurrent_scenario.rs | 4 +-
core/server/src/shard/mod.rs | 5 +-
core/server/src/shard/system/streams.rs | 2 -
core/server/src/shard/system/users.rs | 38 ++++---
core/server/src/slab/users.rs | 121 +++++++++++++++++++++
6 files changed, 169 insertions(+), 19 deletions(-)
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index 5f67c5c06..344b97d55 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -25,7 +25,7 @@ use test_case::test_matrix;
// TODO: Include other trasnsport protocols
#[test_matrix(
- [TransportProtocol::Http, TransportProtocol::Tcp, TransportProtocol::WebSocket],
+ [quic(), tcp(), http(), websocket()],
[
system_scenario(),
user_scenario(),
@@ -40,3 +40,19 @@ use test_case::test_matrix;
async fn matrix(transport: TransportProtocol, scenario: ScenarioFn) {
run_scenario(transport, scenario).await;
}
+
+fn quic() -> TransportProtocol {
+ TransportProtocol::Quic
+}
+
+fn tcp() -> TransportProtocol {
+ TransportProtocol::Tcp
+}
+
+fn http() -> TransportProtocol {
+ TransportProtocol::Http
+}
+
+fn websocket() -> TransportProtocol {
+ TransportProtocol::WebSocket
+}
diff --git a/core/integration/tests/server/scenarios/concurrent_scenario.rs b/core/integration/tests/server/scenarios/concurrent_scenario.rs
index 88e76b3fd..e79f42bbc 100644
--- a/core/integration/tests/server/scenarios/concurrent_scenario.rs
+++ b/core/integration/tests/server/scenarios/concurrent_scenario.rs
@@ -21,8 +21,8 @@ use futures::future::join_all;
use iggy::prelude::*;
use integration::test_server::{ClientFactory, login_root};
-const OPERATIONS_COUNT: usize = 20;
-const MULTIPLE_CLIENT_COUNT: usize = 10;
+const OPERATIONS_COUNT: usize = 100;
+const MULTIPLE_CLIENT_COUNT: usize = 40;
const OPERATIONS_PER_CLIENT: usize = OPERATIONS_COUNT / MULTIPLE_CLIENT_COUNT;
const USER_PASSWORD: &str = "secret";
const TEST_STREAM_NAME: &str = "race-test-stream";
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index b0bc3058e..7a2529fab 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -174,7 +174,10 @@ impl IggyShard {
// JWT token cleaner task is spawned inside HTTP server because it needs `AppState`.
- if self.config.quic.enabled {
+ // TODO(hubcio): QUIC doesn't properly work on all shards, especially tests `concurrent` and `system_scenario`.
+ // it's probably related to Endpoint not Cloned between shards, but all shards are creating its own instance.
+ // This way packet CID is invalid. (crypto-related stuff)
+ if self.config.quic.enabled && self.id == 0 {
continuous::spawn_quic_server(self.clone());
}
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 4ce440599..a67318f6c 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -19,12 +19,10 @@
use super::COMPONENT;
use crate::shard::IggyShard;
use crate::slab::traits_ext::{DeleteCell, EntityMarker, InsertCell};
-
use crate::streaming::session::Session;
use crate::streaming::streams::storage2::{create_stream_file_hierarchy, delete_stream_from_disk};
use crate::streaming::streams::{self, stream2};
use error_set::ErrContext;
-
use iggy_common::{Identifier, IggyError};
impl IggyShard {
diff --git a/core/server/src/shard/system/users.rs b/core/server/src/shard/system/users.rs
index 9ce5cb293..ab1c1e40f 100644
--- a/core/server/src/shard/system/users.rs
+++ b/core/server/src/shard/system/users.rs
@@ -114,13 +114,20 @@ impl IggyShard {
pub fn create_user_bypass_auth(
&self,
- _user_id: u32, // Ignored - Slab auto-assigns IDs
+ expected_user_id: u32,
username: &str,
password: &str,
status: UserStatus,
permissions: Option<Permissions>,
) -> Result<(), IggyError> {
- self.create_user_base(username, password, status, permissions)?;
+ let assigned_user_id = self.create_user_base(username, password, status, permissions)?;
+
+ assert_eq!(
+ assigned_user_id as u32, expected_user_id,
+ "User ID mismatch: expected {}, got {}. This indicates shards are out of sync.",
+ expected_user_id, assigned_user_id
+ );
+
Ok(())
}
@@ -226,26 +233,31 @@ impl IggyShard {
username: Option<String>,
status: Option<UserStatus>,
) -> Result<User, IggyError> {
+ let user = self.get_user(user_id)?;
+ let numeric_user_id = Identifier::numeric(user.id).unwrap();
+
if let Some(ref new_username) = username {
- let user = self.get_user(user_id)?;
let existing_user = self.get_user(&new_username.to_owned().try_into()?);
if existing_user.is_ok() && existing_user.unwrap().id != user.id {
error!("User: {new_username} already exists.");
return Err(IggyError::UserAlreadyExists);
}
+
+ self.users.update_username(user_id, new_username.clone())?;
}
- self.users.with_user_mut(user_id, |user| {
- if let Some(username) = username {
- user.username = username;
- }
- if let Some(status) = status {
+ if let Some(status) = status {
+ self.users.with_user_mut(&numeric_user_id, |user| {
user.status = status;
- }
- user.clone()
- }).with_error_context(|error| {
- format!("{COMPONENT} update user (error: {error}) - failed to update user with id: {user_id}")
- })
+ }).with_error_context(|error| {
+ format!("{COMPONENT} update user (error: {error}) - failed to update user with id: {user_id}")
+ })?;
+ }
+
+ self.get_user(&numeric_user_id)
+ .with_error_context(|error| {
+ format!("{COMPONENT} update user (error: {error}) - failed to get updated user with id: {user_id}")
+ })
}
pub fn update_permissions(
diff --git a/core/server/src/slab/users.rs b/core/server/src/slab/users.rs
index 8a6949fa0..434972fc0 100644
--- a/core/server/src/slab/users.rs
+++ b/core/server/src/slab/users.rs
@@ -137,4 +137,125 @@ impl Users {
pub fn get_id_by_username(&self, username: &str) -> Option<usize> {
self.index.borrow().get(username).copied()
}
+
+ pub fn update_username(
+ &self,
+ identifier: &Identifier,
+ new_username: String,
+ ) -> Result<(), IggyError> {
+ let id = match identifier.kind {
+ iggy_common::IdKind::Numeric => identifier.get_u32_value()? as usize,
+ iggy_common::IdKind::String => {
+ let username = identifier.get_string_value()?;
+ let index = self.index.borrow();
+ *index
+ .get(&username)
+ .ok_or_else(|| IggyError::ResourceNotFound(username.to_string()))?
+ }
+ };
+
+ let old_username = {
+ let users = self.users.borrow();
+ let user = users
+ .get(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(identifier.to_string()))?;
+ user.username.clone()
+ };
+
+ if old_username == new_username {
+ return Ok(());
+ }
+
+ tracing::trace!(
+ "Updating username: '{}' → '{}' for user ID: {}",
+ old_username,
+ new_username,
+ id
+ );
+
+ {
+ let mut users = self.users.borrow_mut();
+ let user = users
+ .get_mut(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(identifier.to_string()))?;
+ user.username = new_username.clone();
+ }
+
+ let mut index = self.index.borrow_mut();
+ index.remove(&old_username);
+ index.insert(new_username, id);
+
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy_common::UserStatus;
+
+ #[test]
+ fn test_username_update_should_maintain_index() {
+ let users = Users::new();
+
+ let mut user = User::new(0, "alice", "password", UserStatus::Active, None);
+ user.id = 0;
+ let id = users.insert(user);
+
+ let alice_id = Identifier::named("alice").unwrap();
+ assert!(users.get_by_identifier(&alice_id).unwrap().is_some());
+ assert_eq!(
+ users
+ .get_by_identifier(&alice_id)
+ .unwrap()
+ .unwrap()
+ .username,
+ "alice"
+ );
+
+ users.update_username(&alice_id, "bob".to_string()).unwrap();
+
+ let bob_id = Identifier::named("bob").unwrap();
+ let bob_lookup = users.get_by_identifier(&bob_id);
+
+ assert!(
+ bob_lookup.unwrap().is_some(),
+ "User should be findable by new username 'bob'"
+ );
+
+ let numeric_id = Identifier::numeric(id as u32).unwrap();
+ let user_by_id = users.get_by_identifier(&numeric_id).unwrap().unwrap();
+ assert_eq!(user_by_id.username, "bob");
+ }
+
+ #[test]
+ fn test_insert_updates_index() {
+ let users = Users::new();
+
+ let user = User::new(0, "alice", "password", UserStatus::Active, None);
+ let id = users.insert(user);
+
+ let alice_id = Identifier::named("alice").unwrap();
+ assert!(users.get_by_identifier(&alice_id).unwrap().is_some());
+
+ let numeric_id = Identifier::numeric(id as u32).unwrap();
+ assert!(users.get_by_identifier(&numeric_id).unwrap().is_some());
+ }
+
+ #[test]
+ fn test_remove_updates_index() {
+ let users = Users::new();
+
+ let user = User::new(0, "alice", "password", UserStatus::Active, None);
+ let id = users.insert(user);
+
+ let removed = users.remove(id);
+ assert!(removed.is_some());
+
+ let alice_id = Identifier::named("alice").unwrap();
+ assert!(users.get_by_identifier(&alice_id).unwrap().is_none());
+
+ let numeric_id = Identifier::numeric(id as u32).unwrap();
+ assert!(users.get_by_identifier(&numeric_id).unwrap().is_none());
+ }
}