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/20 09:42:28 UTC
(iggy) branch io_uring_tpc_shard0 updated (035f0d411 -> bfbbf013b)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch io_uring_tpc_shard0
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 035f0d411 feat(io_uring): route certain commands to shard0 to prevent race conditions
new bfbbf013b feat(io_uring): route certain commands to shard0 to prevent race conditions
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 (035f0d411)
\
N -- N -- N refs/heads/io_uring_tpc_shard0 (bfbbf013b)
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/server/concurrent_addition.rs | 13 +++-
.../tests/server/scenarios/concurrent_scenario.rs | 90 ++++++++++++++++++++--
core/server/src/shard/mod.rs | 2 +-
3 files changed, 93 insertions(+), 12 deletions(-)
(iggy) 01/01: feat(io_uring): route certain commands to shard0 to prevent race conditions
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_shard0
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit bfbbf013ba926b0da4476dd24cde13382c753ca5
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Oct 20 10:48:36 2025 +0200
feat(io_uring): route certain commands to shard0 to prevent race conditions
Route CreateStream, CreateTopic, and CreateUser commands to shard0 for │
centralized processing. Non-shard0 shards forward requests to shard0, │
which processes them and synchronously broadcasts events to all shards │
before responding. This prevents race conditions when multiple clients │
concurrently attempt to create the same entities.
---
core/bench/src/benchmarks/common.rs | 3 +-
.../tests/server/concurrent_addition.rs | 20 +-
.../tests/server/scenarios/concurrent_scenario.rs | 235 ++++++++-------------
.../handlers/streams/create_stream_handler.rs | 111 +++++++---
.../binary/handlers/topics/create_topic_handler.rs | 180 +++++++++++-----
.../binary/handlers/users/create_user_handler.rs | 141 +++++++++----
core/server/src/shard/mod.rs | 167 +++++++++++++--
core/server/src/shard/system/messages.rs | 6 +-
core/server/src/shard/transmission/frame.rs | 9 +-
core/server/src/shard/transmission/message.rs | 25 ++-
10 files changed, 583 insertions(+), 314 deletions(-)
diff --git a/core/bench/src/benchmarks/common.rs b/core/bench/src/benchmarks/common.rs
index fb5bc1192..da6878c0d 100644
--- a/core/bench/src/benchmarks/common.rs
+++ b/core/bench/src/benchmarks/common.rs
@@ -89,7 +89,8 @@ pub async fn init_consumer_groups(
);
if let Err(err) = client
.create_consumer_group(&stream_id, &topic_id, &consumer_group_name)
- .await {
+ .await
+ {
error!("Error when creating consumer group {consumer_group_id}: {err}");
}
}
diff --git a/core/integration/tests/server/concurrent_addition.rs b/core/integration/tests/server/concurrent_addition.rs
index de0aa9c14..ed67b93b5 100644
--- a/core/integration/tests/server/concurrent_addition.rs
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -17,7 +17,7 @@
*/
use crate::server::scenarios::concurrent_scenario::{
- self, ClientCount, ResourceType, ScenarioType,
+ self, ResourceType, ScenarioType, barrier_off, barrier_on,
};
use iggy_common::TransportProtocol;
use integration::{
@@ -31,23 +31,23 @@ use test_case::test_matrix;
// Tests all combinations of:
// - Transport: TCP, HTTP, QUIC, WebSocket (4)
// - Resource: User, Stream, Topic (3)
-// - Clients: Single, Multiple (2)
// - Path: Hot (unique names), Cold (duplicate names) (2)
+// - Barrier: On (synchronized), Off (unsynchronized) (2)
// Total: 4 × 3 × 2 × 2 = 48 test cases
#[test_matrix(
[tcp(), http(), quic(), websocket()],
[user(), stream(), topic()],
- [single_client(), multiple_clients()],
- [hot(), cold()]
+ [hot(), cold()],
+ [barrier_on(), barrier_off()]
)]
#[tokio::test]
#[parallel]
async fn matrix(
transport: TransportProtocol,
resource_type: ResourceType,
- client_count: ClientCount,
path_type: ScenarioType,
+ use_barrier: bool,
) {
let mut test_server = TestServer::default();
test_server.start();
@@ -74,7 +74,7 @@ async fn matrix(
}
};
- concurrent_scenario::run(&*client_factory, resource_type, client_count, path_type).await;
+ concurrent_scenario::run(&*client_factory, resource_type, path_type, use_barrier).await;
}
fn tcp() -> TransportProtocol {
@@ -105,14 +105,6 @@ fn topic() -> ResourceType {
ResourceType::Topic
}
-fn single_client() -> ClientCount {
- ClientCount::Single
-}
-
-fn multiple_clients() -> ClientCount {
- ClientCount::Multiple(10)
-}
-
fn hot() -> ScenarioType {
ScenarioType::Hot
}
diff --git a/core/integration/tests/server/scenarios/concurrent_scenario.rs b/core/integration/tests/server/scenarios/concurrent_scenario.rs
index 3b6fc1bdf..37f15634f 100644
--- a/core/integration/tests/server/scenarios/concurrent_scenario.rs
+++ b/core/integration/tests/server/scenarios/concurrent_scenario.rs
@@ -20,6 +20,8 @@ use crate::server::scenarios::create_client;
use futures::future::join_all;
use iggy::prelude::*;
use integration::test_server::{ClientFactory, login_root};
+use std::sync::Arc;
+use tokio::sync::Barrier;
const OPERATIONS_COUNT: usize = 100;
const MULTIPLE_CLIENT_COUNT: usize = 20;
@@ -37,25 +39,27 @@ pub enum ResourceType {
// TODO(hubcio): add Partition
}
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum ClientCount {
- Single,
- Multiple(usize),
-}
-
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScenarioType {
Hot, // Different names - all should succeed
Cold, // Same name - only first should succeed
}
+pub fn barrier_on() -> bool {
+ true
+}
+
+pub fn barrier_off() -> bool {
+ false
+}
+
type OperationResult = Result<(), IggyError>;
pub async fn run(
client_factory: &dyn ClientFactory,
resource_type: ResourceType,
- client_count: ClientCount,
scenario_type: ScenarioType,
+ use_barrier: bool,
) {
let root_client = create_client(client_factory).await;
login_root(&root_client).await;
@@ -65,42 +69,24 @@ pub async fn run(
root_client.create_stream(TEST_STREAM_NAME).await.unwrap();
}
- let results = match (resource_type, client_count, scenario_type) {
- (ResourceType::User, ClientCount::Single, ScenarioType::Hot) => {
- execute_single_client_users_hot(&root_client).await
- }
- (ResourceType::User, ClientCount::Single, ScenarioType::Cold) => {
- execute_single_client_users_cold(&root_client).await
- }
- (ResourceType::User, ClientCount::Multiple(_), ScenarioType::Hot) => {
- execute_multiple_clients_users_hot(client_factory).await
+ let results = match (resource_type, scenario_type) {
+ (ResourceType::User, ScenarioType::Hot) => {
+ execute_multiple_clients_users_hot(client_factory, use_barrier).await
}
- (ResourceType::User, ClientCount::Multiple(_), ScenarioType::Cold) => {
- execute_multiple_clients_users_cold(client_factory).await
+ (ResourceType::User, ScenarioType::Cold) => {
+ execute_multiple_clients_users_cold(client_factory, use_barrier).await
}
- (ResourceType::Stream, ClientCount::Single, ScenarioType::Hot) => {
- execute_single_client_streams_hot(&root_client).await
+ (ResourceType::Stream, ScenarioType::Hot) => {
+ execute_multiple_clients_streams_hot(client_factory, use_barrier).await
}
- (ResourceType::Stream, ClientCount::Single, ScenarioType::Cold) => {
- execute_single_client_streams_cold(&root_client).await
+ (ResourceType::Stream, ScenarioType::Cold) => {
+ execute_multiple_clients_streams_cold(client_factory, use_barrier).await
}
- (ResourceType::Stream, ClientCount::Multiple(_), ScenarioType::Hot) => {
- execute_multiple_clients_streams_hot(client_factory).await
+ (ResourceType::Topic, ScenarioType::Hot) => {
+ execute_multiple_clients_topics_hot(client_factory, use_barrier).await
}
- (ResourceType::Stream, ClientCount::Multiple(_), ScenarioType::Cold) => {
- execute_multiple_clients_streams_cold(client_factory).await
- }
- (ResourceType::Topic, ClientCount::Single, ScenarioType::Hot) => {
- execute_single_client_topics_hot(&root_client).await
- }
- (ResourceType::Topic, ClientCount::Single, ScenarioType::Cold) => {
- execute_single_client_topics_cold(&root_client).await
- }
- (ResourceType::Topic, ClientCount::Multiple(_), ScenarioType::Hot) => {
- execute_multiple_clients_topics_hot(client_factory).await
- }
- (ResourceType::Topic, ClientCount::Multiple(_), ScenarioType::Cold) => {
- execute_multiple_clients_topics_cold(client_factory).await
+ (ResourceType::Topic, ScenarioType::Cold) => {
+ execute_multiple_clients_topics_cold(client_factory, use_barrier).await
}
};
@@ -109,46 +95,27 @@ pub async fn run(
cleanup_resources(&root_client, resource_type).await;
}
-async fn execute_single_client_users_hot(client: &IggyClient) -> Vec<OperationResult> {
- let mut results = Vec::with_capacity(OPERATIONS_COUNT);
-
- for i in 0..OPERATIONS_COUNT {
- let username = format!("race-user-{}", i);
- let result = client
- .create_user(&username, USER_PASSWORD, UserStatus::Active, None)
- .await
- .map(|_| ());
- results.push(result);
- }
-
- results
-}
-
-async fn execute_single_client_users_cold(client: &IggyClient) -> Vec<OperationResult> {
- let mut results = Vec::with_capacity(OPERATIONS_COUNT);
- const DUPLICATE_USER: &str = "race-user-duplicate";
-
- for _ in 0..OPERATIONS_COUNT {
- let result = client
- .create_user(DUPLICATE_USER, USER_PASSWORD, UserStatus::Active, None)
- .await
- .map(|_| ());
- results.push(result);
- }
-
- results
-}
-
async fn execute_multiple_clients_users_hot(
client_factory: &dyn ClientFactory,
+ use_barrier: bool,
) -> Vec<OperationResult> {
let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+ let barrier = if use_barrier {
+ Some(Arc::new(Barrier::new(MULTIPLE_CLIENT_COUNT)))
+ } else {
+ None
+ };
for client_id in 0..MULTIPLE_CLIENT_COUNT {
let client = create_client(client_factory).await;
login_root(&client).await;
+ let barrier_clone = barrier.clone();
let handle = tokio::spawn(async move {
+ if let Some(b) = barrier_clone {
+ b.wait().await;
+ }
+
let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
for i in 0..OPERATIONS_PER_CLIENT {
let username = format!("race-user-{}-{}", client_id, i);
@@ -164,7 +131,6 @@ async fn execute_multiple_clients_users_hot(
handles.push(handle);
}
- // Await all concurrent operations
let all_results = join_all(handles).await;
all_results
.into_iter()
@@ -174,15 +140,26 @@ async fn execute_multiple_clients_users_hot(
async fn execute_multiple_clients_users_cold(
client_factory: &dyn ClientFactory,
+ use_barrier: bool,
) -> Vec<OperationResult> {
let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
const DUPLICATE_USER: &str = "race-user-duplicate";
+ let barrier = if use_barrier {
+ Some(Arc::new(Barrier::new(MULTIPLE_CLIENT_COUNT)))
+ } else {
+ None
+ };
for _ in 0..MULTIPLE_CLIENT_COUNT {
let client = create_client(client_factory).await;
login_root(&client).await;
+ let barrier_clone = barrier.clone();
let handle = tokio::spawn(async move {
+ if let Some(b) = barrier_clone {
+ b.wait().await;
+ }
+
let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
for _ in 0..OPERATIONS_PER_CLIENT {
let result = client
@@ -204,44 +181,27 @@ async fn execute_multiple_clients_users_cold(
.collect()
}
-async fn execute_single_client_streams_hot(client: &IggyClient) -> Vec<OperationResult> {
- let mut results = Vec::with_capacity(OPERATIONS_COUNT);
-
- for i in 0..OPERATIONS_COUNT {
- let stream_name = format!("race-stream-{}", i);
- let result = client.create_stream(&stream_name).await.map(|_| ());
- results.push(result);
- }
-
- results
-}
-
-async fn execute_single_client_streams_cold(client: &IggyClient) -> Vec<OperationResult> {
- let mut results = Vec::with_capacity(OPERATIONS_COUNT);
- const DUPLICATE_STREAM: &str = "race-stream-duplicate";
-
- for _ in 0..OPERATIONS_COUNT {
- let result = client.create_stream(DUPLICATE_STREAM).await.map(|_| ());
- results.push(result);
- }
-
- results
-}
-
-// ============================================================================
-// MULTIPLE CLIENTS - STREAMS
-// ============================================================================
-
async fn execute_multiple_clients_streams_hot(
client_factory: &dyn ClientFactory,
+ use_barrier: bool,
) -> Vec<OperationResult> {
let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+ let barrier = if use_barrier {
+ Some(Arc::new(Barrier::new(MULTIPLE_CLIENT_COUNT)))
+ } else {
+ None
+ };
for client_id in 0..MULTIPLE_CLIENT_COUNT {
let client = create_client(client_factory).await;
login_root(&client).await;
+ let barrier_clone = barrier.clone();
let handle = tokio::spawn(async move {
+ if let Some(b) = barrier_clone {
+ b.wait().await;
+ }
+
let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
for i in 0..OPERATIONS_PER_CLIENT {
let stream_name = format!("race-stream-{}-{}", client_id, i);
@@ -263,15 +223,26 @@ async fn execute_multiple_clients_streams_hot(
async fn execute_multiple_clients_streams_cold(
client_factory: &dyn ClientFactory,
+ use_barrier: bool,
) -> Vec<OperationResult> {
let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
const DUPLICATE_STREAM: &str = "race-stream-duplicate";
+ let barrier = if use_barrier {
+ Some(Arc::new(Barrier::new(MULTIPLE_CLIENT_COUNT)))
+ } else {
+ None
+ };
for _ in 0..MULTIPLE_CLIENT_COUNT {
let client = create_client(client_factory).await;
login_root(&client).await;
+ let barrier_clone = barrier.clone();
let handle = tokio::spawn(async move {
+ if let Some(b) = barrier_clone {
+ b.wait().await;
+ }
+
let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
for _ in 0..OPERATIONS_PER_CLIENT {
let result = client.create_stream(DUPLICATE_STREAM).await.map(|_| ());
@@ -290,64 +261,27 @@ async fn execute_multiple_clients_streams_cold(
.collect()
}
-async fn execute_single_client_topics_hot(client: &IggyClient) -> Vec<OperationResult> {
- let mut results = Vec::with_capacity(OPERATIONS_COUNT);
- let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
-
- for i in 0..OPERATIONS_COUNT {
- let topic_name = format!("race-topic-{}", i);
- let result = client
- .create_topic(
- &stream_id,
- &topic_name,
- PARTITIONS_COUNT,
- CompressionAlgorithm::default(),
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .map(|_| ());
- results.push(result);
- }
-
- results
-}
-
-async fn execute_single_client_topics_cold(client: &IggyClient) -> Vec<OperationResult> {
- let mut results = Vec::with_capacity(OPERATIONS_COUNT);
- const DUPLICATE_TOPIC: &str = "race-topic-duplicate";
- let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
-
- for _ in 0..OPERATIONS_COUNT {
- let result = client
- .create_topic(
- &stream_id,
- DUPLICATE_TOPIC,
- PARTITIONS_COUNT,
- CompressionAlgorithm::default(),
- None,
- IggyExpiry::NeverExpire,
- MaxTopicSize::ServerDefault,
- )
- .await
- .map(|_| ());
- results.push(result);
- }
-
- results
-}
-
async fn execute_multiple_clients_topics_hot(
client_factory: &dyn ClientFactory,
+ use_barrier: bool,
) -> Vec<OperationResult> {
let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+ let barrier = if use_barrier {
+ Some(Arc::new(Barrier::new(MULTIPLE_CLIENT_COUNT)))
+ } else {
+ None
+ };
for client_id in 0..MULTIPLE_CLIENT_COUNT {
let client = create_client(client_factory).await;
login_root(&client).await;
+ let barrier_clone = barrier.clone();
let handle = tokio::spawn(async move {
+ if let Some(b) = barrier_clone {
+ b.wait().await;
+ }
+
let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
@@ -382,15 +316,26 @@ async fn execute_multiple_clients_topics_hot(
async fn execute_multiple_clients_topics_cold(
client_factory: &dyn ClientFactory,
+ use_barrier: bool,
) -> Vec<OperationResult> {
let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
const DUPLICATE_TOPIC: &str = "race-topic-duplicate";
+ let barrier = if use_barrier {
+ Some(Arc::new(Barrier::new(MULTIPLE_CLIENT_COUNT)))
+ } else {
+ None
+ };
for _ in 0..MULTIPLE_CLIENT_COUNT {
let client = create_client(client_factory).await;
login_root(&client).await;
+ let barrier_clone = barrier.clone();
let handle = tokio::spawn(async move {
+ if let Some(b) = barrier_clone {
+ b.wait().await;
+ }
+
let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
diff --git a/core/server/src/binary/handlers/streams/create_stream_handler.rs b/core/server/src/binary/handlers/streams/create_stream_handler.rs
index cfabf569d..227cbe747 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -23,15 +23,18 @@ use crate::binary::{handlers::streams::COMPONENT, sender::SenderKind};
use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
-use crate::shard_info;
+use crate::shard::transmission::frame::ShardResponse;
+use crate::shard::transmission::message::{
+ ShardMessage, ShardRequest, ShardRequestPayload, ShardSendRequestResult,
+};
use crate::slab::traits_ext::{EntityComponentSystem, EntityMarker};
use crate::state::command::EntryCommand;
use crate::state::models::CreateStreamWithId;
use crate::streaming::session::Session;
use anyhow::Result;
use error_set::ErrContext;
-use iggy_common::IggyError;
use iggy_common::create_stream::CreateStream;
+use iggy_common::{Identifier, IggyError};
use std::rc::Rc;
use tracing::{debug, instrument};
@@ -49,39 +52,81 @@ impl ServerCommandHandler for CreateStream {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let stream = shard.create_stream2(session, self.name.clone()).await?;
- let created_stream_id = stream.id();
- shard_info!(
- shard.id,
- "Created stream with ID: {}, name: '{}'.",
- created_stream_id,
- stream.root().name()
- );
- let event = ShardEvent::CreatedStream2 {
- id: created_stream_id,
- stream,
- };
- shard.broadcast_event_to_all_shards(event).await?;
+ if shard.id != 0 {
+ let request = ShardRequest {
+ stream_id: Identifier::default(),
+ topic_id: Identifier::default(),
+ partition_id: 0,
+ payload: ShardRequestPayload::CreateStream {
+ user_id: session.get_user_id(),
+ name: self.name.clone(),
+ },
+ };
+
+ let message = ShardMessage::Request(request);
+ match shard.send_request_to_shard_or_recoil(None, message).await? {
+ ShardSendRequestResult::Response(ShardResponse::CreateStreamResponse(stream)) => {
+ let created_stream_id = stream.id();
+ let response = mapper::map_stream(stream.root(), stream.stats());
+
+ shard
+ .state
+ .apply(session.get_user_id(), &EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id: created_stream_id as u32,
+ command: self
+ }))
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create stream for id: {created_stream_id}, session: {session}"
+ )
+ })?;
+
+ sender.send_ok_response(&response).await?;
+ }
+ ShardSendRequestResult::Response(ShardResponse::ErrorResponse(err)) => {
+ return Err(err);
+ }
+ ShardSendRequestResult::Recoil(_) => {
+ unreachable!("Unexpected recoil when forwarding CreateStream to shard0");
+ }
+ _ => {
+ return Err(IggyError::InvalidCommand);
+ }
+ }
+ } else {
+ let stream = shard.create_stream2(session, self.name.clone()).await?;
+ let created_stream_id = stream.id();
+
+ let event = ShardEvent::CreatedStream2 {
+ id: created_stream_id,
+ stream: stream.clone(),
+ };
+ shard.broadcast_event_to_all_shards(event).await?;
+
+ let response = shard
+ .streams2
+ .with_components_by_id(created_stream_id, |(root, stats)| {
+ mapper::map_stream(&root, &stats)
+ });
+
+ shard
+ .state
+ .apply(session.get_user_id(), &EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id: created_stream_id as u32,
+ command: self
+ }))
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create stream for id: {created_stream_id}, session: {session}"
+ )
+ })?;
+
+ sender.send_ok_response(&response).await?;
+ }
- let response = shard
- .streams2
- .with_components_by_id(created_stream_id, |(root, stats)| {
- mapper::map_stream(&root, &stats)
- });
- shard
- .state
- .apply(session.get_user_id(), &EntryCommand::CreateStream(CreateStreamWithId {
- stream_id: created_stream_id as u32,
- command: self
- }))
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to apply create stream for id: {created_stream_id}, session: {session}"
- )
- })?;
- sender.send_ok_response(&response).await?;
Ok(())
}
}
diff --git a/core/server/src/binary/handlers/topics/create_topic_handler.rs b/core/server/src/binary/handlers/topics/create_topic_handler.rs
index fadb137cd..72998f372 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -23,6 +23,10 @@ use crate::binary::{handlers::topics::COMPONENT, sender::SenderKind};
use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
+use crate::shard::transmission::frame::ShardResponse;
+use crate::shard::transmission::message::{
+ ShardMessage, ShardRequest, ShardRequestPayload, ShardSendRequestResult,
+};
use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::state::models::CreateTopicWithId;
@@ -49,64 +53,128 @@ impl ServerCommandHandler for CreateTopic {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let topic = shard
- .create_topic2(
- session,
- &self.stream_id,
- self.name.clone(),
- self.message_expiry,
- self.compression_algorithm,
- self.max_topic_size,
- self.replication_factor,
- )
- .await?;
- self.message_expiry = topic.root().message_expiry();
- self.max_topic_size = topic.root().max_topic_size();
-
- let stream_id = shard
- .streams2
- .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
- let topic_id = topic.id();
- // Send events for topic creation.
- let event = ShardEvent::CreatedTopic2 {
- stream_id: self.stream_id.clone(),
- topic,
- };
-
- shard.broadcast_event_to_all_shards(event).await?;
- let partitions = shard
- .create_partitions2(
- session,
+
+ if shard.id != 0 {
+ let request = ShardRequest {
+ stream_id: Identifier::default(),
+ topic_id: Identifier::default(),
+ partition_id: 0,
+ payload: ShardRequestPayload::CreateTopic {
+ user_id: session.get_user_id(),
+ stream_id: self.stream_id.clone(),
+ name: self.name.clone(),
+ partitions_count: self.partitions_count,
+ message_expiry: self.message_expiry,
+ compression_algorithm: self.compression_algorithm,
+ max_topic_size: self.max_topic_size,
+ replication_factor: self.replication_factor,
+ },
+ };
+
+ let message = ShardMessage::Request(request);
+ match shard.send_request_to_shard_or_recoil(None, message).await? {
+ ShardSendRequestResult::Response(ShardResponse::CreateTopicResponse(topic)) => {
+ let topic_id = topic.id();
+ self.message_expiry = topic.root().message_expiry();
+ self.max_topic_size = topic.root().max_topic_size();
+
+ let stream_id = shard
+ .streams2
+ .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
+
+ let response = shard.streams2.with_topic_by_id(
+ &self.stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ |(root, _, stats)| mapper::map_topic(&root, &stats),
+ );
+
+ shard
+ .state
+ .apply(session.get_user_id(), &EntryCommand::CreateTopic(CreateTopicWithId {
+ topic_id: topic_id as u32,
+ command: self
+ }))
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create topic for stream_id: {stream_id}, topic_id: {topic_id:?}"
+ )
+ })?;
+
+ sender.send_ok_response(&response).await?;
+ }
+ ShardSendRequestResult::Response(ShardResponse::ErrorResponse(err)) => {
+ return Err(err);
+ }
+ ShardSendRequestResult::Recoil(_) => {
+ unreachable!("Unexpected recoil when forwarding CreateTopic to shard0");
+ }
+ _ => {
+ return Err(IggyError::InvalidCommand);
+ }
+ }
+ } else {
+ let topic = shard
+ .create_topic2(
+ session,
+ &self.stream_id,
+ self.name.clone(),
+ self.message_expiry,
+ self.compression_algorithm,
+ self.max_topic_size,
+ self.replication_factor,
+ )
+ .await?;
+ self.message_expiry = topic.root().message_expiry();
+ self.max_topic_size = topic.root().max_topic_size();
+
+ let stream_id = shard
+ .streams2
+ .with_stream_by_id(&self.stream_id, streams::helpers::get_stream_id());
+ let topic_id = topic.id();
+
+ let event = ShardEvent::CreatedTopic2 {
+ stream_id: self.stream_id.clone(),
+ topic,
+ };
+
+ shard.broadcast_event_to_all_shards(event).await?;
+ let partitions = shard
+ .create_partitions2(
+ session,
+ &self.stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ self.partitions_count,
+ )
+ .await?;
+ let event = ShardEvent::CreatedPartitions2 {
+ stream_id: self.stream_id.clone(),
+ topic_id: Identifier::numeric(topic_id as u32).unwrap(),
+ partitions,
+ };
+
+ shard.broadcast_event_to_all_shards(event).await?;
+
+ let response = shard.streams2.with_topic_by_id(
&self.stream_id,
&Identifier::numeric(topic_id as u32).unwrap(),
- self.partitions_count,
- )
- .await?;
- let event = ShardEvent::CreatedPartitions2 {
- stream_id: self.stream_id.clone(),
- topic_id: Identifier::numeric(topic_id as u32).unwrap(),
- partitions,
- };
-
- shard.broadcast_event_to_all_shards(event).await?;
- let response = shard.streams2.with_topic_by_id(
- &self.stream_id,
- &Identifier::numeric(topic_id as u32).unwrap(),
- |(root, _, stats)| mapper::map_topic(&root, &stats),
- );
- shard
- .state
- .apply(session.get_user_id(), &EntryCommand::CreateTopic(CreateTopicWithId {
- topic_id: topic_id as u32,
- command: self
- }))
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to apply create topic for stream_id: {stream_id}, topic_id: {topic_id:?}"
- )
- })?;
- sender.send_ok_response(&response).await?;
+ |(root, _, stats)| mapper::map_topic(&root, &stats),
+ );
+ shard
+ .state
+ .apply(session.get_user_id(), &EntryCommand::CreateTopic(CreateTopicWithId {
+ topic_id: topic_id as u32,
+ command: self
+ }))
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create topic for stream_id: {stream_id}, topic_id: {topic_id:?}"
+ )
+ })?;
+ sender.send_ok_response(&response).await?;
+ }
+
Ok(())
}
}
diff --git a/core/server/src/binary/handlers/users/create_user_handler.rs b/core/server/src/binary/handlers/users/create_user_handler.rs
index 8dc18e0f7..ef35fa207 100644
--- a/core/server/src/binary/handlers/users/create_user_handler.rs
+++ b/core/server/src/binary/handlers/users/create_user_handler.rs
@@ -18,7 +18,11 @@
use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
-use crate::{shard_debug, shard_info};
+use crate::shard::transmission::frame::ShardResponse;
+use crate::shard::transmission::message::{
+ ShardMessage, ShardRequest, ShardRequestPayload, ShardSendRequestResult,
+};
+use crate::shard_debug;
use std::rc::Rc;
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
@@ -31,8 +35,8 @@ use crate::streaming::session::Session;
use crate::streaming::utils::crypto;
use anyhow::Result;
use error_set::ErrContext;
-use iggy_common::IggyError;
use iggy_common::create_user::CreateUser;
+use iggy_common::{Identifier, IggyError};
use tracing::instrument;
impl ServerCommandHandler for CreateUser {
@@ -50,8 +54,62 @@ impl ServerCommandHandler for CreateUser {
) -> Result<(), IggyError> {
shard_debug!(shard.id, "session: {session}, command: {self}");
- shard_info!(shard.id, "Creating user: {}", self.username);
- let user = shard
+ if shard.id != 0 {
+ let request = ShardRequest {
+ stream_id: Identifier::default(),
+ topic_id: Identifier::default(),
+ partition_id: 0,
+ payload: ShardRequestPayload::CreateUser {
+ user_id: session.get_user_id(),
+ username: self.username.clone(),
+ password: self.password.clone(),
+ status: self.status,
+ permissions: self.permissions.clone(),
+ },
+ };
+
+ let message = ShardMessage::Request(request);
+ match shard.send_request_to_shard_or_recoil(None, message).await? {
+ ShardSendRequestResult::Response(ShardResponse::CreateUserResponse(user)) => {
+ let user_id = user.id;
+
+ let response = mapper::map_user(&user);
+
+ shard
+ .state
+ .apply(
+ session.get_user_id(),
+ &EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: CreateUser {
+ username: self.username.to_owned(),
+ password: crypto::hash_password(&self.password),
+ status: self.status,
+ permissions: self.permissions.clone(),
+ },
+ }),
+ )
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create user for user_id: {user_id}, session: {session}"
+ )
+ })?;
+
+ sender.send_ok_response(&response).await?;
+ }
+ ShardSendRequestResult::Response(ShardResponse::ErrorResponse(err)) => {
+ return Err(err);
+ }
+ ShardSendRequestResult::Recoil(_) => {
+ unreachable!("Unexpected recoil when forwarding CreateUser to shard0");
+ }
+ _ => {
+ return Err(IggyError::InvalidCommand);
+ }
+ }
+ } else {
+ let user = shard
.create_user(
session,
&self.username,
@@ -65,46 +123,45 @@ impl ServerCommandHandler for CreateUser {
self.username, session
)
})?;
- shard_info!(
- shard.id,
- "Created user: {} with ID: {}.",
- user.username,
- user.id
- );
- let event = ShardEvent::CreatedUser {
- user_id: user.id,
- username: self.username.to_owned(),
- password: self.password.to_owned(),
- status: self.status,
- permissions: self.permissions.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
- let user_id = user.id;
- let response = mapper::map_user(&user);
-
- // For the security of the system, we hash the password before storing it in metadata.
- shard
- .state
- .apply(
- session.get_user_id(),
- &EntryCommand::CreateUser(CreateUserWithId {
+
+ let user_id = user.id;
+
+ let event = ShardEvent::CreatedUser {
user_id,
- command: CreateUser {
- username: self.username.to_owned(),
- password: crypto::hash_password(&self.password),
- status: self.status,
- permissions: self.permissions.clone(),
- }
- }),
- )
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to apply create user with name: {}, session: {session}",
- self.username
+ username: self.username.to_owned(),
+ password: self.password.to_owned(),
+ status: self.status,
+ permissions: self.permissions.clone(),
+ };
+ shard.broadcast_event_to_all_shards(event).await?;
+
+ let response = mapper::map_user(&user);
+
+ shard
+ .state
+ .apply(
+ session.get_user_id(),
+ &EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: CreateUser {
+ username: self.username.to_owned(),
+ password: crypto::hash_password(&self.password),
+ status: self.status,
+ permissions: self.permissions.clone(),
+ },
+ }),
)
- })?;
- sender.send_ok_response(&response).await?;
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create user with name: {}, session: {session}",
+ self.username
+ )
+ })?;
+
+ sender.send_ok_response(&response).await?;
+ }
+
Ok(())
}
}
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 9abf6835c..68ca71191 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -65,7 +65,7 @@ use transmission::connector::{Receiver, ShardConnector, StopReceiver};
pub const COMPONENT: &str = "SHARD";
pub const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
-pub const BROADCAST_TIMEOUT: Duration = Duration::from_secs(20);
+pub const BROADCAST_TIMEOUT: Duration = Duration::from_secs(60);
pub(crate) struct Shard {
id: u16,
@@ -516,6 +516,120 @@ impl IggyShard {
.await?;
Ok(ShardResponse::FlushUnsavedBuffer)
}
+ ShardRequestPayload::CreateStream { user_id, name } => {
+ assert_eq!(self.id, 0, "CreateStream should only be handled by shard0");
+
+ let session = Session::stateless(
+ user_id,
+ std::net::SocketAddr::new(
+ std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
+ 0,
+ ),
+ );
+
+ let stream = self.create_stream2(&session, name.clone()).await?;
+ let created_stream_id = stream.id();
+
+ let event = ShardEvent::CreatedStream2 {
+ id: created_stream_id,
+ stream: stream.clone(),
+ };
+
+ self.broadcast_event_to_all_shards(event).await?;
+
+ Ok(ShardResponse::CreateStreamResponse(stream))
+ }
+ ShardRequestPayload::CreateTopic {
+ user_id,
+ stream_id,
+ name,
+ partitions_count,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ replication_factor,
+ } => {
+ assert_eq!(self.id, 0, "CreateTopic should only be handled by shard0");
+
+ let session = Session::stateless(
+ user_id,
+ std::net::SocketAddr::new(
+ std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
+ 0,
+ ),
+ );
+
+ let topic = self
+ .create_topic2(
+ &session,
+ &stream_id,
+ name.clone(),
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ replication_factor,
+ )
+ .await?;
+
+ let topic_id = topic.id();
+
+ let event = ShardEvent::CreatedTopic2 {
+ stream_id: stream_id.clone(),
+ topic: topic.clone(),
+ };
+ self.broadcast_event_to_all_shards(event).await?;
+
+ let partitions = self
+ .create_partitions2(
+ &session,
+ &stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ partitions_count,
+ )
+ .await?;
+
+ let event = ShardEvent::CreatedPartitions2 {
+ stream_id: stream_id.clone(),
+ topic_id: Identifier::numeric(topic_id as u32).unwrap(),
+ partitions,
+ };
+ self.broadcast_event_to_all_shards(event).await?;
+
+ Ok(ShardResponse::CreateTopicResponse(topic))
+ }
+ ShardRequestPayload::CreateUser {
+ user_id,
+ username,
+ password,
+ status,
+ permissions,
+ } => {
+ assert_eq!(self.id, 0, "CreateUser should only be handled by shard0");
+
+ let session = Session::stateless(
+ user_id,
+ std::net::SocketAddr::new(
+ std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
+ 0,
+ ),
+ );
+
+ let user =
+ self.create_user(&session, &username, &password, status, permissions.clone())?;
+
+ let created_user_id = user.id;
+
+ let event = ShardEvent::CreatedUser {
+ user_id: created_user_id,
+ username: username.clone(),
+ password: password.clone(),
+ status,
+ permissions: permissions.clone(),
+ };
+ self.broadcast_event_to_all_shards(event).await?;
+
+ Ok(ShardResponse::CreateUserResponse(user))
+ }
}
}
@@ -871,31 +985,47 @@ impl IggyShard {
pub async fn send_request_to_shard_or_recoil(
&self,
- namespace: &IggyNamespace,
+ namespace: Option<&IggyNamespace>,
message: ShardMessage,
) -> Result<ShardSendRequestResult, IggyError> {
- if let Some(shard) = self.find_shard(namespace) {
- if shard.id == self.id {
+ if let Some(ns) = namespace {
+ if let Some(shard) = self.find_shard(ns) {
+ if shard.id == self.id {
+ return Ok(ShardSendRequestResult::Recoil(message));
+ }
+
+ let response = match shard.send_request(message).await {
+ Ok(response) => response,
+ Err(err) => {
+ error!(
+ "{COMPONENT} - failed to send request to shard with ID: {}, error: {err}",
+ shard.id
+ );
+ return Err(err);
+ }
+ };
+ Ok(ShardSendRequestResult::Response(response))
+ } else {
+ Err(IggyError::ShardNotFound(
+ ns.stream_id(),
+ ns.topic_id(),
+ ns.partition_id(),
+ ))
+ }
+ } else {
+ if self.id == 0 {
return Ok(ShardSendRequestResult::Recoil(message));
}
- let response = match shard.send_request(message).await {
+ let shard0 = &self.shards[0];
+ let response = match shard0.send_request(message).await {
Ok(response) => response,
Err(err) => {
- error!(
- "{COMPONENT} - failed to send request to shard with ID: {}, error: {err}",
- shard.id
- );
+ error!("{COMPONENT} - failed to send admin request to shard0, error: {err}");
return Err(err);
}
};
Ok(ShardSendRequestResult::Response(response))
- } else {
- Err(IggyError::ShardNotFound(
- namespace.stream_id(),
- namespace.topic_id(),
- namespace.partition_id(),
- ))
}
}
@@ -935,12 +1065,13 @@ impl IggyShard {
);
Err(())
}
- Err(_) => {
+ Err(e) => {
shard_warn!(
self_id,
- "Broadcast to shard {} failed for event {}: timeout waiting for response after {:?}",
+ "Broadcast to shard {} failed for event {}: timeout waiting for response after {:?}, elapsed: {:?}",
shard_id, event_type,
- BROADCAST_TIMEOUT
+ BROADCAST_TIMEOUT,
+ e
);
Err(())
}
diff --git a/core/server/src/shard/system/messages.rs b/core/server/src/shard/system/messages.rs
index 967adb058..0e398df09 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -105,7 +105,7 @@ impl IggyShard {
let request = ShardRequest::new(stream_id.clone(), topic_id.clone(), partition_id, payload);
let message = ShardMessage::Request(request);
match self
- .send_request_to_shard_or_recoil(&namespace, message)
+ .send_request_to_shard_or_recoil(Some(&namespace), message)
.await?
{
ShardSendRequestResult::Recoil(message) => {
@@ -214,7 +214,7 @@ impl IggyShard {
let request = ShardRequest::new(stream_id.clone(), topic_id.clone(), partition_id, payload);
let message = ShardMessage::Request(request);
let (metadata, batch) = match self
- .send_request_to_shard_or_recoil(&namespace, message)
+ .send_request_to_shard_or_recoil(Some(&namespace), message)
.await?
{
ShardSendRequestResult::Recoil(message) => {
@@ -306,7 +306,7 @@ impl IggyShard {
let request = ShardRequest::new(stream_id.clone(), topic_id.clone(), partition_id, payload);
let message = ShardMessage::Request(request);
match self
- .send_request_to_shard_or_recoil(&namespace, message)
+ .send_request_to_shard_or_recoil(Some(&namespace), message)
.await?
{
ShardSendRequestResult::Recoil(message) => {
diff --git a/core/server/src/shard/transmission/frame.rs b/core/server/src/shard/transmission/frame.rs
index d36b92157..e5555644a 100644
--- a/core/server/src/shard/transmission/frame.rs
+++ b/core/server/src/shard/transmission/frame.rs
@@ -20,15 +20,22 @@ use iggy_common::IggyError;
use crate::{
binary::handlers::messages::poll_messages_handler::IggyPollMetadata,
- shard::transmission::message::ShardMessage, streaming::segments::IggyMessagesBatchSet,
+ shard::transmission::message::ShardMessage,
+ streaming::{
+ segments::IggyMessagesBatchSet, streams::stream2, topics::topic2, users::user::User,
+ },
};
+#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
pub enum ShardResponse {
PollMessages((IggyPollMetadata, IggyMessagesBatchSet)),
SendMessages,
FlushUnsavedBuffer,
Event,
+ CreateStreamResponse(stream2::Stream),
+ CreateTopicResponse(topic2::Topic),
+ CreateUserResponse(User),
ErrorResponse(IggyError),
}
diff --git a/core/server/src/shard/transmission/message.rs b/core/server/src/shard/transmission/message.rs
index 5a5d650e6..038c460ea 100644
--- a/core/server/src/shard/transmission/message.rs
+++ b/core/server/src/shard/transmission/message.rs
@@ -23,7 +23,9 @@ use crate::{
slab::partitions,
streaming::{polling_consumer::PollingConsumer, segments::IggyMessagesBatchMut},
};
-use iggy_common::Identifier;
+use iggy_common::{
+ CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, Permissions, UserStatus,
+};
#[allow(clippy::large_enum_variant)]
pub enum ShardSendRequestResult {
@@ -75,6 +77,27 @@ pub enum ShardRequestPayload {
FlushUnsavedBuffer {
fsync: bool,
},
+ CreateStream {
+ user_id: u32,
+ name: String,
+ },
+ CreateTopic {
+ user_id: u32,
+ stream_id: Identifier,
+ name: String,
+ partitions_count: u32,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: Option<u8>,
+ },
+ CreateUser {
+ user_id: u32,
+ username: String,
+ password: String,
+ status: UserStatus,
+ permissions: Option<Permissions>,
+ },
}
impl From<ShardRequest> for ShardMessage {