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/12/29 15:37:13 UTC
(iggy) branch arc-swap created (now c7fe0e2c8)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch arc-swap
in repository https://gitbox.apache.org/repos/asf/iggy.git
at c7fe0e2c8 XD
This branch includes the following new commits:
new b01b96e41 metadata
new 0d9fab57e cg, users, partitions
new 058ae50ce commands
new c7fe0e2c8 XD
The 4 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/04: metadata
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch arc-swap
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b01b96e41461979ebc225cb854bd9d48c961bce9
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Dec 29 14:06:27 2025 +0100
metadata
---
Cargo.lock | 8 +-
DEPENDENCIES.md | 2 +-
core/server/Cargo.toml | 1 +
core/server/src/lib.rs | 1 +
core/server/src/main.rs | 8 +
core/server/src/metadata/consumer_group.rs | 35 ++
core/server/src/metadata/mod.rs | 38 ++
core/server/src/metadata/partition.rs | 32 ++
core/server/src/metadata/shared.rs | 836 +++++++++++++++++++++++++++++
core/server/src/metadata/snapshot.rs | 182 +++++++
core/server/src/metadata/stream.rs | 80 +++
core/server/src/metadata/topic.rs | 113 ++++
core/server/src/metadata/user.rs | 70 +++
core/server/src/shard/builder.rs | 9 +
core/server/src/shard/handlers.rs | 19 +-
core/server/src/shard/mod.rs | 4 +
core/server/src/shard/system/streams.rs | 15 +-
core/server/src/shard/system/topics.rs | 28 +-
18 files changed, 1465 insertions(+), 16 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 0f671da6d..5df8c8f3c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -426,9 +426,12 @@ dependencies = [
[[package]]
name = "arc-swap"
-version = "1.7.1"
+version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
+checksum = "51d03449bb8ca2cc2ef70869af31463d1ae5ccc8fa3e334b307203fbf815207e"
+dependencies = [
+ "rustversion",
+]
[[package]]
name = "arcshift"
@@ -8199,6 +8202,7 @@ version = "0.6.1-edge.1"
dependencies = [
"ahash 0.8.12",
"anyhow",
+ "arc-swap",
"arcshift",
"argon2",
"async-channel",
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index 8b06b77ac..72ca495bd 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -32,7 +32,7 @@ anstyle-wincon: 3.0.11, "Apache-2.0 OR MIT",
anyhow: 1.0.100, "Apache-2.0 OR MIT",
apache-avro: 0.17.0, "Apache-2.0",
arbitrary: 1.4.2, "Apache-2.0 OR MIT",
-arc-swap: 1.7.1, "Apache-2.0 OR MIT",
+arc-swap: 1.8.0, "Apache-2.0 OR MIT",
arcshift: 0.4.2, "Apache-2.0 OR MIT",
argon2: 0.5.3, "Apache-2.0 OR MIT",
array-init: 2.1.0, "Apache-2.0 OR MIT",
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index bbf15794a..6e0d6d6d0 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -40,6 +40,7 @@ iggy-web = ["dep:rust-embed", "dep:mime_guess"]
[dependencies]
ahash = { workspace = true }
anyhow = { workspace = true }
+arc-swap = "1.8.0"
arcshift = "0.4.2"
argon2 = { workspace = true }
async-channel = { workspace = true }
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index 2919214b6..41669486f 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -37,6 +37,7 @@ pub mod diagnostics;
pub mod http;
pub mod io;
pub mod log;
+pub mod metadata;
pub mod quic;
pub mod server_error;
pub mod shard;
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 58c50e33c..51b7f0566 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -34,6 +34,7 @@ use server::configs::sharding::ShardAllocator;
use server::diagnostics::{print_io_uring_permission_info, print_locked_memory_limit_info};
use server::io::fs_utils;
use server::log::logger::Logging;
+use server::metadata::SharedMetadata;
use server::server_error::ServerError;
use server::shard::namespace::IggyNamespace;
use server::shard::system::info::SystemInfo;
@@ -321,6 +322,11 @@ fn main() -> Result<(), ServerError> {
let client_manager: EternalPtr<DashMap<u32, Client>> = client_manager.into();
let client_manager = ClientManager::new(client_manager);
+ // Create shared metadata for all shards (ArcSwap-based lock-free reads)
+ let shared_metadata = Box::new(SharedMetadata::new());
+ let shared_metadata = Box::leak(shared_metadata);
+ let shared_metadata: EternalPtr<SharedMetadata> = shared_metadata.into();
+
streams.with_components(|components| {
let (root, ..) = components.into_components();
for (_, stream) in root.iter() {
@@ -353,6 +359,7 @@ fn main() -> Result<(), ServerError> {
{
let streams = streams.clone();
let shards_table = shards_table.clone();
+ let shared_metadata = shared_metadata.clone();
let users = users.clone();
let connections = connections.clone();
let config = config.clone();
@@ -417,6 +424,7 @@ fn main() -> Result<(), ServerError> {
.state(state)
.users(users)
.shards_table(shards_table)
+ .shared_metadata(shared_metadata)
.connections(connections)
.clients_manager(client_manager)
.config(config)
diff --git a/core/server/src/metadata/consumer_group.rs b/core/server/src/metadata/consumer_group.rs
new file mode 100644
index 000000000..34c1d20b5
--- /dev/null
+++ b/core/server/src/metadata/consumer_group.rs
@@ -0,0 +1,35 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/// Consumer group metadata stored in the shared snapshot.
+/// Member management is dynamic and handled separately.
+#[derive(Debug, Clone)]
+pub struct ConsumerGroupMeta {
+ pub id: usize,
+ pub name: String,
+ pub partitions: Vec<usize>,
+}
+
+impl ConsumerGroupMeta {
+ pub fn new(id: usize, name: String, partitions: Vec<usize>) -> Self {
+ Self {
+ id,
+ name,
+ partitions,
+ }
+ }
+}
diff --git a/core/server/src/metadata/mod.rs b/core/server/src/metadata/mod.rs
new file mode 100644
index 000000000..e964dccd9
--- /dev/null
+++ b/core/server/src/metadata/mod.rs
@@ -0,0 +1,38 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Shared metadata module providing a single source of truth for all shards.
+//!
+//! This module replaces the broadcast-based metadata synchronization with an
+//! `ArcSwap`-based approach where all shards read from a shared snapshot,
+//! and only shard 0 can write (swap in new snapshots).
+
+mod consumer_group;
+mod partition;
+mod shared;
+mod snapshot;
+mod stream;
+mod topic;
+mod user;
+
+pub use consumer_group::ConsumerGroupMeta;
+pub use partition::PartitionMeta;
+pub use shared::SharedMetadata;
+pub use snapshot::MetadataSnapshot;
+pub use stream::StreamMeta;
+pub use topic::TopicMeta;
+pub use user::{PersonalAccessTokenMeta, UserMeta};
diff --git a/core/server/src/metadata/partition.rs b/core/server/src/metadata/partition.rs
new file mode 100644
index 000000000..57e50dfa9
--- /dev/null
+++ b/core/server/src/metadata/partition.rs
@@ -0,0 +1,32 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use iggy_common::IggyTimestamp;
+
+/// Partition metadata stored in the shared snapshot.
+/// The actual partition data (segments, journals) is stored per-shard.
+#[derive(Debug, Clone)]
+pub struct PartitionMeta {
+ pub id: usize,
+ pub created_at: IggyTimestamp,
+}
+
+impl PartitionMeta {
+ pub fn new(id: usize, created_at: IggyTimestamp) -> Self {
+ Self { id, created_at }
+ }
+}
diff --git a/core/server/src/metadata/shared.rs b/core/server/src/metadata/shared.rs
new file mode 100644
index 000000000..f03086b65
--- /dev/null
+++ b/core/server/src/metadata/shared.rs
@@ -0,0 +1,836 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::metadata::{
+ ConsumerGroupMeta, MetadataSnapshot, PartitionMeta, StreamMeta, TopicMeta, UserMeta,
+};
+use arc_swap::{ArcSwap, Guard};
+use iggy_common::{
+ CompressionAlgorithm, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize,
+ Permissions, UserStatus,
+};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
+
+/// Shared metadata accessible by all shards.
+///
+/// Uses ArcSwap for lock-free reads and atomic updates.
+/// Only shard 0 should call write methods (create/update/delete).
+/// All shards can call read methods (get/exists/load).
+pub struct SharedMetadata {
+ inner: ArcSwap<MetadataSnapshot>,
+
+ // Atomic ID generators (only shard 0 increments these)
+ next_stream_id: AtomicUsize,
+ next_user_id: AtomicU32,
+}
+
+impl Default for SharedMetadata {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl SharedMetadata {
+ pub fn new() -> Self {
+ Self {
+ inner: ArcSwap::from_pointee(MetadataSnapshot::new()),
+ next_stream_id: AtomicUsize::new(0),
+ next_user_id: AtomicU32::new(0),
+ }
+ }
+
+ /// Initialize with existing state (used during server startup).
+ pub fn init_from_snapshot(
+ &self,
+ snapshot: MetadataSnapshot,
+ next_stream_id: usize,
+ next_user_id: u32,
+ ) {
+ self.inner.store(Arc::new(snapshot));
+ self.next_stream_id.store(next_stream_id, Ordering::SeqCst);
+ self.next_user_id.store(next_user_id, Ordering::SeqCst);
+ }
+
+ // ==================== Read Operations (any shard) ====================
+
+ /// Load the current snapshot. Lock-free read.
+ pub fn load(&self) -> Guard<Arc<MetadataSnapshot>> {
+ self.inner.load()
+ }
+
+ /// Get a clone of the current snapshot.
+ pub fn snapshot(&self) -> Arc<MetadataSnapshot> {
+ Arc::clone(&self.inner.load())
+ }
+
+ // Stream reads
+
+ pub fn stream_exists(&self, id: &Identifier) -> bool {
+ self.inner.load().stream_exists(id)
+ }
+
+ pub fn stream_exists_by_name(&self, name: &str) -> bool {
+ self.inner.load().stream_exists_by_name(name)
+ }
+
+ pub fn get_stream(&self, id: &Identifier) -> Option<StreamMeta> {
+ self.inner.load().get_stream(id).cloned()
+ }
+
+ pub fn get_stream_id(&self, id: &Identifier) -> Option<usize> {
+ self.inner.load().get_stream_id(id)
+ }
+
+ pub fn get_streams(&self) -> Vec<StreamMeta> {
+ self.inner.load().streams.values().cloned().collect()
+ }
+
+ // User reads
+
+ pub fn user_exists(&self, id: &Identifier) -> bool {
+ self.inner.load().user_exists(id)
+ }
+
+ pub fn user_exists_by_name(&self, username: &str) -> bool {
+ self.inner.load().user_exists_by_name(username)
+ }
+
+ pub fn get_user(&self, id: &Identifier) -> Option<UserMeta> {
+ self.inner.load().get_user(id).cloned()
+ }
+
+ pub fn get_user_id(&self, id: &Identifier) -> Option<u32> {
+ self.inner.load().get_user_id(id)
+ }
+
+ pub fn get_users(&self) -> Vec<UserMeta> {
+ self.inner.load().users.values().cloned().collect()
+ }
+
+ pub fn get_user_by_username(&self, username: &str) -> Option<UserMeta> {
+ self.inner.load().get_user_by_username(username).cloned()
+ }
+
+ // ==================== Write Operations (shard 0 only) ====================
+
+ // Stream operations
+
+ /// Create a new stream. Returns the created stream metadata.
+ pub fn create_stream(&self, name: String) -> Result<StreamMeta, IggyError> {
+ let current = self.inner.load();
+
+ if current.stream_exists_by_name(&name) {
+ return Err(IggyError::StreamNameAlreadyExists(name));
+ }
+
+ let id = self.next_stream_id.fetch_add(1, Ordering::SeqCst);
+ let stream = StreamMeta::new(id, name.clone(), IggyTimestamp::now());
+
+ let mut new_snapshot = (**current).clone();
+ new_snapshot.stream_index.insert(name, id);
+ new_snapshot.streams.insert(id, stream.clone());
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(stream)
+ }
+
+ /// Delete a stream. Returns the deleted stream metadata.
+ pub fn delete_stream(&self, id: &Identifier) -> Result<StreamMeta, IggyError> {
+ let current = self.inner.load();
+
+ let stream_id = current
+ .get_stream_id(id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(id.clone()))?;
+
+ let mut new_snapshot = (**current).clone();
+ let stream = new_snapshot
+ .streams
+ .remove(&stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(id.clone()))?;
+ new_snapshot.stream_index.remove(&stream.name);
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(stream)
+ }
+
+ /// Update a stream's name.
+ pub fn update_stream(&self, id: &Identifier, new_name: String) -> Result<(), IggyError> {
+ let current = self.inner.load();
+
+ let stream_id = current
+ .get_stream_id(id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(id.clone()))?;
+
+ let stream = current
+ .streams
+ .get(&stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(id.clone()))?;
+
+ if stream.name == new_name {
+ return Ok(());
+ }
+
+ if current.stream_exists_by_name(&new_name) {
+ return Err(IggyError::StreamNameAlreadyExists(new_name));
+ }
+
+ let mut new_snapshot = (**current).clone();
+ let old_name = new_snapshot.streams.get(&stream_id).unwrap().name.clone();
+ new_snapshot.stream_index.remove(&old_name);
+ new_snapshot
+ .stream_index
+ .insert(new_name.clone(), stream_id);
+ new_snapshot.streams.get_mut(&stream_id).unwrap().name = new_name;
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(())
+ }
+
+ // Topic operations
+
+ /// Create a new topic in a stream.
+ #[allow(clippy::too_many_arguments)]
+ pub fn create_topic(
+ &self,
+ stream_id: &Identifier,
+ name: String,
+ replication_factor: u8,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ ) -> Result<TopicMeta, IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ if stream.topic_exists(&name) {
+ return Err(IggyError::TopicNameAlreadyExists(name, stream_id.clone()));
+ }
+
+ // Topic IDs are sequential within a stream
+ let topic_id = stream.topics.len();
+ let topic = TopicMeta::new(
+ topic_id,
+ name.clone(),
+ IggyTimestamp::now(),
+ replication_factor,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ );
+
+ let mut new_snapshot = (**current).clone();
+ new_snapshot
+ .streams
+ .get_mut(&stream_idx)
+ .unwrap()
+ .add_topic(topic.clone());
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(topic)
+ }
+
+ /// Delete a topic from a stream.
+ pub fn delete_topic(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ ) -> Result<TopicMeta, IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let topic_idx = match topic_id.kind {
+ iggy_common::IdKind::Numeric => topic_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let name = topic_id.get_string_value().unwrap();
+ stream.get_topic_id_by_name(&name).ok_or_else(|| {
+ IggyError::TopicNameNotFound(name.to_string(), stream.name.clone())
+ })?
+ }
+ };
+
+ let mut new_snapshot = (**current).clone();
+ let topic = new_snapshot
+ .streams
+ .get_mut(&stream_idx)
+ .unwrap()
+ .remove_topic(topic_idx)
+ .ok_or_else(|| IggyError::TopicIdNotFound(topic_id.clone(), stream_id.clone()))?;
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(topic)
+ }
+
+ /// Update topic configuration.
+ #[allow(clippy::too_many_arguments)]
+ pub fn update_topic(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ name: Option<String>,
+ message_expiry: Option<IggyExpiry>,
+ compression_algorithm: Option<CompressionAlgorithm>,
+ max_topic_size: Option<MaxTopicSize>,
+ replication_factor: Option<u8>,
+ ) -> Result<(), IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let stream_name = stream.name.clone();
+ let topic_idx = match topic_id.kind {
+ iggy_common::IdKind::Numeric => topic_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let topic_name = topic_id.get_string_value().unwrap();
+ stream.get_topic_id_by_name(&topic_name).ok_or_else(|| {
+ IggyError::TopicNameNotFound(topic_name.to_string(), stream_name.clone())
+ })?
+ }
+ };
+
+ let mut new_snapshot = (**current).clone();
+ let stream_mut = new_snapshot.streams.get_mut(&stream_idx).unwrap();
+
+ // Check name availability before getting mutable topic reference
+ if let Some(ref new_name) = name {
+ let old_name = stream_mut
+ .topics
+ .get(&topic_idx)
+ .ok_or_else(|| IggyError::TopicIdNotFound(topic_id.clone(), stream_id.clone()))?
+ .name
+ .clone();
+ if *new_name != old_name && stream_mut.topic_exists(new_name) {
+ return Err(IggyError::TopicNameAlreadyExists(
+ new_name.clone(),
+ stream_id.clone(),
+ ));
+ }
+ }
+
+ let topic = stream_mut
+ .topics
+ .get_mut(&topic_idx)
+ .ok_or_else(|| IggyError::TopicIdNotFound(topic_id.clone(), stream_id.clone()))?;
+
+ if let Some(new_name) = name
+ && topic.name != new_name
+ {
+ stream_mut.topic_index.remove(&topic.name);
+ stream_mut.topic_index.insert(new_name.clone(), topic_idx);
+ topic.name = new_name;
+ }
+ if let Some(expiry) = message_expiry {
+ topic.message_expiry = expiry;
+ }
+ if let Some(compression) = compression_algorithm {
+ topic.compression_algorithm = compression;
+ }
+ if let Some(size) = max_topic_size {
+ topic.max_topic_size = size;
+ }
+ if let Some(factor) = replication_factor {
+ topic.replication_factor = factor;
+ }
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(())
+ }
+
+ // Partition operations
+
+ /// Add partitions to a topic.
+ pub fn create_partitions(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ count: u32,
+ ) -> Result<Vec<PartitionMeta>, IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let topic_idx = match topic_id.kind {
+ iggy_common::IdKind::Numeric => topic_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let name = topic_id.get_string_value().unwrap();
+ stream.get_topic_id_by_name(&name).ok_or_else(|| {
+ IggyError::TopicNameNotFound(name.to_string(), stream.name.clone())
+ })?
+ }
+ };
+
+ let topic = stream.topics.get(&topic_idx).unwrap();
+ let start_id = topic.partitions.len();
+
+ let mut new_partitions = Vec::with_capacity(count as usize);
+ for i in 0..count {
+ let partition_id = start_id + i as usize;
+ new_partitions.push(PartitionMeta::new(partition_id, IggyTimestamp::now()));
+ }
+
+ let mut new_snapshot = (**current).clone();
+ let topic_mut = new_snapshot
+ .streams
+ .get_mut(&stream_idx)
+ .unwrap()
+ .topics
+ .get_mut(&topic_idx)
+ .unwrap();
+
+ for partition in &new_partitions {
+ topic_mut.add_partition(partition.clone());
+ }
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(new_partitions)
+ }
+
+ /// Delete partitions from a topic.
+ pub fn delete_partitions(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition_ids: &[usize],
+ ) -> Result<Vec<PartitionMeta>, IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let topic_idx = match topic_id.kind {
+ iggy_common::IdKind::Numeric => topic_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let name = topic_id.get_string_value().unwrap();
+ stream.get_topic_id_by_name(&name).ok_or_else(|| {
+ IggyError::TopicNameNotFound(name.to_string(), stream.name.clone())
+ })?
+ }
+ };
+
+ let mut new_snapshot = (**current).clone();
+ let topic = new_snapshot
+ .streams
+ .get_mut(&stream_idx)
+ .unwrap()
+ .topics
+ .get_mut(&topic_idx)
+ .unwrap();
+
+ let mut deleted = Vec::with_capacity(partition_ids.len());
+ for &id in partition_ids {
+ if let Some(partition) = topic.remove_partition(id) {
+ deleted.push(partition);
+ }
+ }
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(deleted)
+ }
+
+ // User operations
+
+ /// Create a new user.
+ pub fn create_user(
+ &self,
+ username: String,
+ password_hash: String,
+ status: UserStatus,
+ permissions: Option<Permissions>,
+ ) -> Result<UserMeta, IggyError> {
+ let current = self.inner.load();
+
+ if current.user_exists_by_name(&username) {
+ return Err(IggyError::UserAlreadyExists);
+ }
+
+ let id = self.next_user_id.fetch_add(1, Ordering::SeqCst);
+ let user = UserMeta::new(
+ id,
+ username.clone(),
+ password_hash,
+ IggyTimestamp::now(),
+ status,
+ permissions,
+ );
+
+ let mut new_snapshot = (**current).clone();
+ new_snapshot.user_index.insert(username, id);
+ new_snapshot.users.insert(id, user.clone());
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(user)
+ }
+
+ /// Delete a user.
+ pub fn delete_user(&self, id: &Identifier) -> Result<UserMeta, IggyError> {
+ let current = self.inner.load();
+
+ let user_id = current
+ .get_user_id(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {id}")))?;
+
+ let mut new_snapshot = (**current).clone();
+ let user = new_snapshot
+ .users
+ .remove(&user_id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {user_id}")))?;
+ new_snapshot.user_index.remove(&user.username);
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(user)
+ }
+
+ /// Update user metadata.
+ pub fn update_user(
+ &self,
+ id: &Identifier,
+ username: Option<String>,
+ status: Option<UserStatus>,
+ ) -> Result<(), IggyError> {
+ let current = self.inner.load();
+
+ let user_id = current
+ .get_user_id(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {id}")))?;
+
+ let mut new_snapshot = (**current).clone();
+ let user = new_snapshot
+ .users
+ .get_mut(&user_id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {user_id}")))?;
+
+ if let Some(new_username) = username
+ && user.username != new_username
+ {
+ if new_snapshot.user_index.contains_key(&new_username) {
+ return Err(IggyError::UserAlreadyExists);
+ }
+ new_snapshot.user_index.remove(&user.username);
+ new_snapshot
+ .user_index
+ .insert(new_username.clone(), user_id);
+ user.username = new_username;
+ }
+ if let Some(new_status) = status {
+ user.status = new_status;
+ }
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(())
+ }
+
+ /// Update user permissions.
+ pub fn update_permissions(
+ &self,
+ id: &Identifier,
+ permissions: Option<Permissions>,
+ ) -> Result<(), IggyError> {
+ let current = self.inner.load();
+
+ let user_id = current
+ .get_user_id(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {id}")))?;
+
+ let mut new_snapshot = (**current).clone();
+ let user = new_snapshot
+ .users
+ .get_mut(&user_id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {user_id}")))?;
+ user.permissions = permissions;
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(())
+ }
+
+ /// Change user password.
+ pub fn change_password(
+ &self,
+ id: &Identifier,
+ new_password_hash: String,
+ ) -> Result<(), IggyError> {
+ let current = self.inner.load();
+
+ let user_id = current
+ .get_user_id(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {id}")))?;
+
+ let mut new_snapshot = (**current).clone();
+ let user = new_snapshot
+ .users
+ .get_mut(&user_id)
+ .ok_or_else(|| IggyError::ResourceNotFound(format!("user with ID {user_id}")))?;
+ user.password_hash = new_password_hash;
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(())
+ }
+
+ // Consumer group operations
+
+ /// Create a consumer group.
+ pub fn create_consumer_group(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ name: String,
+ partition_ids: Vec<usize>,
+ ) -> Result<ConsumerGroupMeta, IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let topic_idx = match topic_id.kind {
+ iggy_common::IdKind::Numeric => topic_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let topic_name = topic_id.get_string_value().unwrap();
+ stream.get_topic_id_by_name(&topic_name).ok_or_else(|| {
+ IggyError::TopicNameNotFound(topic_name.to_string(), stream.name.clone())
+ })?
+ }
+ };
+
+ let topic = stream.topics.get(&topic_idx).unwrap();
+ if topic.get_consumer_group_id_by_name(&name).is_some() {
+ return Err(IggyError::ConsumerGroupNameAlreadyExists(
+ name,
+ topic_id.clone(),
+ ));
+ }
+
+ let group_id = topic.consumer_groups.len();
+ let group = ConsumerGroupMeta::new(group_id, name, partition_ids);
+
+ let mut new_snapshot = (**current).clone();
+ new_snapshot
+ .streams
+ .get_mut(&stream_idx)
+ .unwrap()
+ .topics
+ .get_mut(&topic_idx)
+ .unwrap()
+ .add_consumer_group(group.clone());
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(group)
+ }
+
+ /// Delete a consumer group.
+ pub fn delete_consumer_group(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ group_id: &Identifier,
+ ) -> Result<ConsumerGroupMeta, IggyError> {
+ let current = self.inner.load();
+
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let topic_idx = match topic_id.kind {
+ iggy_common::IdKind::Numeric => topic_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let topic_name = topic_id.get_string_value().unwrap();
+ stream.get_topic_id_by_name(&topic_name).ok_or_else(|| {
+ IggyError::TopicNameNotFound(topic_name.to_string(), stream.name.clone())
+ })?
+ }
+ };
+
+ let topic = stream.topics.get(&topic_idx).unwrap();
+ let group_idx = match group_id.kind {
+ iggy_common::IdKind::Numeric => group_id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let group_name = group_id.get_string_value().unwrap();
+ topic
+ .get_consumer_group_id_by_name(&group_name)
+ .ok_or_else(|| {
+ IggyError::ConsumerGroupNameNotFound(
+ group_name.to_string(),
+ topic_id.clone(),
+ )
+ })?
+ }
+ };
+
+ let mut new_snapshot = (**current).clone();
+ let group = new_snapshot
+ .streams
+ .get_mut(&stream_idx)
+ .unwrap()
+ .topics
+ .get_mut(&topic_idx)
+ .unwrap()
+ .remove_consumer_group(group_idx)
+ .ok_or_else(|| {
+ IggyError::ConsumerGroupIdNotFound(group_id.clone(), topic_id.clone())
+ })?;
+
+ self.inner.store(Arc::new(new_snapshot));
+
+ Ok(group)
+ }
+
+ // Bound address operations
+
+ /// Set the bound TCP address.
+ pub fn set_tcp_address(&self, address: SocketAddr) {
+ let current = self.inner.load();
+ let mut new_snapshot = (**current).clone();
+ new_snapshot.bound_addresses.tcp = Some(address);
+ self.inner.store(Arc::new(new_snapshot));
+ }
+
+ /// Set the bound HTTP address.
+ pub fn set_http_address(&self, address: SocketAddr) {
+ let current = self.inner.load();
+ let mut new_snapshot = (**current).clone();
+ new_snapshot.bound_addresses.http = Some(address);
+ self.inner.store(Arc::new(new_snapshot));
+ }
+
+ /// Set the bound QUIC address.
+ pub fn set_quic_address(&self, address: SocketAddr) {
+ let current = self.inner.load();
+ let mut new_snapshot = (**current).clone();
+ new_snapshot.bound_addresses.quic = Some(address);
+ self.inner.store(Arc::new(new_snapshot));
+ }
+
+ /// Set the bound WebSocket address.
+ pub fn set_websocket_address(&self, address: SocketAddr) {
+ let current = self.inner.load();
+ let mut new_snapshot = (**current).clone();
+ new_snapshot.bound_addresses.websocket = Some(address);
+ self.inner.store(Arc::new(new_snapshot));
+ }
+
+ /// Get the current next stream ID (for testing/debugging).
+ pub fn next_stream_id(&self) -> usize {
+ self.next_stream_id.load(Ordering::SeqCst)
+ }
+
+ /// Get the current next user ID (for testing/debugging).
+ pub fn next_user_id(&self) -> u32 {
+ self.next_user_id.load(Ordering::SeqCst)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_create_stream() {
+ let metadata = SharedMetadata::new();
+
+ let stream = metadata.create_stream("test-stream".to_string()).unwrap();
+ assert_eq!(stream.id, 0);
+ assert_eq!(stream.name, "test-stream");
+
+ assert!(metadata.stream_exists_by_name("test-stream"));
+ assert!(!metadata.stream_exists_by_name("nonexistent"));
+ }
+
+ #[test]
+ fn test_create_duplicate_stream_fails() {
+ let metadata = SharedMetadata::new();
+
+ metadata.create_stream("test-stream".to_string()).unwrap();
+ let result = metadata.create_stream("test-stream".to_string());
+
+ assert!(matches!(result, Err(IggyError::StreamNameAlreadyExists(_))));
+ }
+
+ #[test]
+ fn test_delete_stream() {
+ let metadata = SharedMetadata::new();
+
+ let stream = metadata.create_stream("test-stream".to_string()).unwrap();
+ let stream_id = Identifier::numeric(stream.id as u32).unwrap();
+
+ let deleted = metadata.delete_stream(&stream_id).unwrap();
+ assert_eq!(deleted.id, stream.id);
+ assert!(!metadata.stream_exists_by_name("test-stream"));
+ }
+
+ #[test]
+ fn test_create_user() {
+ let metadata = SharedMetadata::new();
+
+ let user = metadata
+ .create_user(
+ "testuser".to_string(),
+ "hashed_password".to_string(),
+ UserStatus::Active,
+ None,
+ )
+ .unwrap();
+
+ assert_eq!(user.id, 0);
+ assert_eq!(user.username, "testuser");
+ assert!(metadata.user_exists_by_name("testuser"));
+ }
+
+ #[test]
+ fn test_concurrent_stream_ids() {
+ let metadata = SharedMetadata::new();
+
+ let stream1 = metadata.create_stream("stream-1".to_string()).unwrap();
+ let stream2 = metadata.create_stream("stream-2".to_string()).unwrap();
+ let stream3 = metadata.create_stream("stream-3".to_string()).unwrap();
+
+ assert_eq!(stream1.id, 0);
+ assert_eq!(stream2.id, 1);
+ assert_eq!(stream3.id, 2);
+ }
+}
diff --git a/core/server/src/metadata/snapshot.rs b/core/server/src/metadata/snapshot.rs
new file mode 100644
index 000000000..8283740a2
--- /dev/null
+++ b/core/server/src/metadata/snapshot.rs
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::metadata::{StreamMeta, UserMeta};
+use ahash::AHashMap;
+use iggy_common::Identifier;
+use std::net::SocketAddr;
+
+/// Bound addresses for transport protocols.
+#[derive(Debug, Clone, Default)]
+pub struct BoundAddresses {
+ pub tcp: Option<SocketAddr>,
+ pub http: Option<SocketAddr>,
+ pub quic: Option<SocketAddr>,
+ pub websocket: Option<SocketAddr>,
+}
+
+/// Immutable metadata snapshot that is atomically swapped.
+/// All reads see a consistent view of the entire metadata.
+#[derive(Debug, Clone)]
+pub struct MetadataSnapshot {
+ /// Stream metadata indexed by stream ID
+ pub streams: AHashMap<usize, StreamMeta>,
+
+ /// Stream name to ID index
+ pub stream_index: AHashMap<String, usize>,
+
+ /// User metadata indexed by user ID
+ pub users: AHashMap<u32, UserMeta>,
+
+ /// Username to user ID index
+ pub user_index: AHashMap<String, u32>,
+
+ /// Bound addresses for transport protocols
+ pub bound_addresses: BoundAddresses,
+}
+
+impl Default for MetadataSnapshot {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl MetadataSnapshot {
+ pub fn new() -> Self {
+ Self {
+ streams: AHashMap::new(),
+ stream_index: AHashMap::new(),
+ users: AHashMap::new(),
+ user_index: AHashMap::new(),
+ bound_addresses: BoundAddresses::default(),
+ }
+ }
+
+ // Stream operations
+
+ pub fn stream_exists_by_name(&self, name: &str) -> bool {
+ self.stream_index.contains_key(name)
+ }
+
+ pub fn stream_exists(&self, id: &Identifier) -> bool {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = id.get_u32_value().unwrap() as usize;
+ self.streams.contains_key(&id)
+ }
+ iggy_common::IdKind::String => {
+ let name = id.get_string_value().unwrap();
+ self.stream_index.contains_key(&name)
+ }
+ }
+ }
+
+ pub fn get_stream(&self, id: &Identifier) -> Option<&StreamMeta> {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = id.get_u32_value().unwrap() as usize;
+ self.streams.get(&id)
+ }
+ iggy_common::IdKind::String => {
+ let name = id.get_string_value().unwrap();
+ self.stream_index
+ .get(&name)
+ .and_then(|id| self.streams.get(id))
+ }
+ }
+ }
+
+ pub fn get_stream_id(&self, id: &Identifier) -> Option<usize> {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let stream_id = id.get_u32_value().unwrap() as usize;
+ if self.streams.contains_key(&stream_id) {
+ Some(stream_id)
+ } else {
+ None
+ }
+ }
+ iggy_common::IdKind::String => {
+ let name = id.get_string_value().unwrap();
+ self.stream_index.get(&name).copied()
+ }
+ }
+ }
+
+ pub fn get_streams(&self) -> Vec<&StreamMeta> {
+ self.streams.values().collect()
+ }
+
+ // User operations
+
+ pub fn user_exists_by_name(&self, username: &str) -> bool {
+ self.user_index.contains_key(username)
+ }
+
+ pub fn user_exists(&self, id: &Identifier) -> bool {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = id.get_u32_value().unwrap();
+ self.users.contains_key(&id)
+ }
+ iggy_common::IdKind::String => {
+ let name = id.get_string_value().unwrap();
+ self.user_index.contains_key(&name)
+ }
+ }
+ }
+
+ pub fn get_user(&self, id: &Identifier) -> Option<&UserMeta> {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = id.get_u32_value().unwrap();
+ self.users.get(&id)
+ }
+ iggy_common::IdKind::String => {
+ let name = id.get_string_value().unwrap();
+ self.user_index.get(&name).and_then(|id| self.users.get(id))
+ }
+ }
+ }
+
+ pub fn get_user_id(&self, id: &Identifier) -> Option<u32> {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let user_id = id.get_u32_value().unwrap();
+ if self.users.contains_key(&user_id) {
+ Some(user_id)
+ } else {
+ None
+ }
+ }
+ iggy_common::IdKind::String => {
+ let name = id.get_string_value().unwrap();
+ self.user_index.get(&name).copied()
+ }
+ }
+ }
+
+ pub fn get_users(&self) -> Vec<&UserMeta> {
+ self.users.values().collect()
+ }
+
+ pub fn get_user_by_username(&self, username: &str) -> Option<&UserMeta> {
+ self.user_index
+ .get(username)
+ .and_then(|id| self.users.get(id))
+ }
+}
diff --git a/core/server/src/metadata/stream.rs b/core/server/src/metadata/stream.rs
new file mode 100644
index 000000000..9b3495630
--- /dev/null
+++ b/core/server/src/metadata/stream.rs
@@ -0,0 +1,80 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::metadata::TopicMeta;
+use ahash::AHashMap;
+use iggy_common::IggyTimestamp;
+
+/// Stream metadata stored in the shared snapshot.
+#[derive(Debug, Clone)]
+pub struct StreamMeta {
+ pub id: usize,
+ pub name: String,
+ pub created_at: IggyTimestamp,
+
+ /// Topic metadata indexed by topic ID
+ pub topics: AHashMap<usize, TopicMeta>,
+
+ /// Topic name to ID index
+ pub topic_index: AHashMap<String, usize>,
+}
+
+impl StreamMeta {
+ pub fn new(id: usize, name: String, created_at: IggyTimestamp) -> Self {
+ Self {
+ id,
+ name,
+ created_at,
+ topics: AHashMap::new(),
+ topic_index: AHashMap::new(),
+ }
+ }
+
+ pub fn topics_count(&self) -> usize {
+ self.topics.len()
+ }
+
+ pub fn add_topic(&mut self, topic: TopicMeta) {
+ self.topic_index.insert(topic.name.clone(), topic.id);
+ self.topics.insert(topic.id, topic);
+ }
+
+ pub fn remove_topic(&mut self, topic_id: usize) -> Option<TopicMeta> {
+ if let Some(topic) = self.topics.remove(&topic_id) {
+ self.topic_index.remove(&topic.name);
+ Some(topic)
+ } else {
+ None
+ }
+ }
+
+ pub fn get_topic(&self, topic_id: usize) -> Option<&TopicMeta> {
+ self.topics.get(&topic_id)
+ }
+
+ pub fn get_topic_mut(&mut self, topic_id: usize) -> Option<&mut TopicMeta> {
+ self.topics.get_mut(&topic_id)
+ }
+
+ pub fn get_topic_id_by_name(&self, name: &str) -> Option<usize> {
+ self.topic_index.get(name).copied()
+ }
+
+ pub fn topic_exists(&self, name: &str) -> bool {
+ self.topic_index.contains_key(name)
+ }
+}
diff --git a/core/server/src/metadata/topic.rs b/core/server/src/metadata/topic.rs
new file mode 100644
index 000000000..f69b42e06
--- /dev/null
+++ b/core/server/src/metadata/topic.rs
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::metadata::{ConsumerGroupMeta, PartitionMeta};
+use ahash::AHashMap;
+use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize};
+
+/// Topic metadata stored in the shared snapshot.
+#[derive(Debug, Clone)]
+pub struct TopicMeta {
+ pub id: usize,
+ pub name: String,
+ pub created_at: IggyTimestamp,
+ pub replication_factor: u8,
+ pub message_expiry: IggyExpiry,
+ pub compression_algorithm: CompressionAlgorithm,
+ pub max_topic_size: MaxTopicSize,
+
+ /// Partition metadata indexed by partition ID
+ pub partitions: AHashMap<usize, PartitionMeta>,
+
+ /// Consumer group metadata indexed by group ID
+ pub consumer_groups: AHashMap<usize, ConsumerGroupMeta>,
+
+ /// Consumer group name to ID index
+ pub consumer_group_index: AHashMap<String, usize>,
+
+ /// Next partition ID for round-robin assignment (not cloned, use atomic)
+ next_partition_id: usize,
+}
+
+impl TopicMeta {
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ id: usize,
+ name: String,
+ created_at: IggyTimestamp,
+ replication_factor: u8,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ ) -> Self {
+ Self {
+ id,
+ name,
+ created_at,
+ replication_factor,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ partitions: AHashMap::new(),
+ consumer_groups: AHashMap::new(),
+ consumer_group_index: AHashMap::new(),
+ next_partition_id: 0,
+ }
+ }
+
+ pub fn partitions_count(&self) -> u32 {
+ self.partitions.len() as u32
+ }
+
+ pub fn add_partition(&mut self, partition: PartitionMeta) {
+ self.partitions.insert(partition.id, partition);
+ }
+
+ pub fn remove_partition(&mut self, partition_id: usize) -> Option<PartitionMeta> {
+ self.partitions.remove(&partition_id)
+ }
+
+ pub fn add_consumer_group(&mut self, group: ConsumerGroupMeta) {
+ self.consumer_group_index
+ .insert(group.name.clone(), group.id);
+ self.consumer_groups.insert(group.id, group);
+ }
+
+ pub fn remove_consumer_group(&mut self, group_id: usize) -> Option<ConsumerGroupMeta> {
+ if let Some(group) = self.consumer_groups.remove(&group_id) {
+ self.consumer_group_index.remove(&group.name);
+ Some(group)
+ } else {
+ None
+ }
+ }
+
+ pub fn get_consumer_group_id_by_name(&self, name: &str) -> Option<usize> {
+ self.consumer_group_index.get(name).copied()
+ }
+
+ pub fn get_next_partition_id(&self, upperbound: usize) -> usize {
+ if upperbound == 0 {
+ return 0;
+ }
+ self.next_partition_id % upperbound
+ }
+
+ pub fn increment_partition_counter(&mut self) {
+ self.next_partition_id = self.next_partition_id.wrapping_add(1);
+ }
+}
diff --git a/core/server/src/metadata/user.rs b/core/server/src/metadata/user.rs
new file mode 100644
index 000000000..d75bd80e2
--- /dev/null
+++ b/core/server/src/metadata/user.rs
@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use ahash::AHashMap;
+use iggy_common::{IggyTimestamp, Permissions, UserStatus};
+
+/// Personal access token metadata.
+#[derive(Debug, Clone)]
+pub struct PersonalAccessTokenMeta {
+ pub name: String,
+ pub token_hash: String,
+ pub expiry_at: Option<IggyTimestamp>,
+}
+
+impl PersonalAccessTokenMeta {
+ pub fn new(name: String, token_hash: String, expiry_at: Option<IggyTimestamp>) -> Self {
+ Self {
+ name,
+ token_hash,
+ expiry_at,
+ }
+ }
+}
+
+/// User metadata stored in the shared snapshot.
+#[derive(Debug, Clone)]
+pub struct UserMeta {
+ pub id: u32,
+ pub username: String,
+ pub password_hash: String,
+ pub created_at: IggyTimestamp,
+ pub status: UserStatus,
+ pub permissions: Option<Permissions>,
+ pub personal_access_tokens: AHashMap<String, PersonalAccessTokenMeta>,
+}
+
+impl UserMeta {
+ pub fn new(
+ id: u32,
+ username: String,
+ password_hash: String,
+ created_at: IggyTimestamp,
+ status: UserStatus,
+ permissions: Option<Permissions>,
+ ) -> Self {
+ Self {
+ id,
+ username,
+ password_hash,
+ created_at,
+ status,
+ permissions,
+ personal_access_tokens: AHashMap::new(),
+ }
+ }
+}
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 188452fe8..e394b9e0c 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -22,6 +22,7 @@ use super::{
};
use crate::{
configs::server::ServerConfig,
+ metadata::SharedMetadata,
shard::namespace::IggyNamespace,
slab::{streams::Streams, users::Users},
state::file::FileState,
@@ -40,6 +41,7 @@ pub struct IggyShardBuilder {
id: Option<u16>,
streams: Option<Streams>,
shards_table: Option<EternalPtr<DashMap<IggyNamespace, ShardId>>>,
+ shared_metadata: Option<EternalPtr<SharedMetadata>>,
state: Option<FileState>,
users: Option<Users>,
client_manager: Option<ClientManager>,
@@ -75,6 +77,11 @@ impl IggyShardBuilder {
self
}
+ pub fn shared_metadata(mut self, shared_metadata: EternalPtr<SharedMetadata>) -> Self {
+ self.shared_metadata = Some(shared_metadata);
+ self
+ }
+
pub fn clients_manager(mut self, client_manager: ClientManager) -> Self {
self.client_manager = Some(client_manager);
self
@@ -120,6 +127,7 @@ impl IggyShardBuilder {
let id = self.id.unwrap();
let streams = self.streams.unwrap();
let shards_table = self.shards_table.unwrap();
+ let shared_metadata = self.shared_metadata.unwrap();
let state = self.state.unwrap();
let users = self.users.unwrap();
let config = self.config.unwrap();
@@ -154,6 +162,7 @@ impl IggyShardBuilder {
id,
shards,
shards_table,
+ shared_metadata,
streams, // TODO: Fixme
users,
fs_locks: Default::default(),
diff --git a/core/server/src/shard/handlers.rs b/core/server/src/shard/handlers.rs
index 087b7a54d..81c54a98a 100644
--- a/core/server/src/shard/handlers.rs
+++ b/core/server/src/shard/handlers.rs
@@ -509,9 +509,10 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
}
Ok(())
}
- ShardEvent::CreatedStream { id, stream } => {
- let stream_id = shard.create_stream_bypass_auth(stream);
- assert_eq!(stream_id, id);
+ ShardEvent::CreatedStream { id: _, stream } => {
+ // Note: Slab::insert() returns the next available slot, not the stream's embedded ID.
+ // The stream already has the correct ID from shard 0, which is preserved after insert.
+ shard.create_stream_bypass_auth(stream);
Ok(())
}
ShardEvent::DeletedStream { id, stream_id } => {
@@ -521,9 +522,9 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
Ok(())
}
ShardEvent::CreatedTopic { stream_id, topic } => {
- let topic_id_from_event = topic.id();
- let topic_id = shard.create_topic_bypass_auth(&stream_id, topic.clone());
- assert_eq!(topic_id, topic_id_from_event);
+ // Note: Slab::insert() returns the next available slot, not the topic's embedded ID.
+ // The topic already has the correct ID from shard 0, which is preserved after insert.
+ shard.create_topic_bypass_auth(&stream_id, topic.clone());
Ok(())
}
ShardEvent::CreatedPartitions {
@@ -570,9 +571,9 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
topic_id,
cg,
} => {
- let cg_id = cg.id();
- let id = shard.create_consumer_group_bypass_auth(&stream_id, &topic_id, cg);
- assert_eq!(id, cg_id);
+ // Note: Slab::insert() returns the next available slot, not the consumer group's embedded ID.
+ // The consumer group already has the correct ID from shard 0.
+ shard.create_consumer_group_bypass_auth(&stream_id, &topic_id, cg);
Ok(())
}
ShardEvent::DeletedConsumerGroup {
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index f48279362..b5424cd9f 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -33,6 +33,7 @@ use self::tasks::{continuous, periodic};
use crate::{
configs::server::ServerConfig,
io::fs_locks::FsLocks,
+ metadata::SharedMetadata,
shard::{
namespace::IggyNamespace, task_registry::TaskRegistry, transmission::frame::ShardFrame,
},
@@ -73,6 +74,9 @@ pub struct IggyShard {
pub(crate) shards_table: EternalPtr<DashMap<IggyNamespace, ShardId>>,
pub(crate) state: FileState,
+ /// Shared metadata accessible by all shards (ArcSwap-based).
+ pub(crate) shared_metadata: EternalPtr<SharedMetadata>,
+
pub(crate) fs_locks: FsLocks,
pub(crate) encryptor: Option<EncryptorKind>,
pub(crate) config: ServerConfig,
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 84b82e0e4..b12ca2f2f 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -42,8 +42,12 @@ impl IggyShard {
if exists {
return Err(IggyError::StreamNameAlreadyExists(name));
}
- let stream = stream::create_and_insert_stream_mem(&self.streams, name);
+ let stream = stream::create_and_insert_stream_mem(&self.streams, name.clone());
self.metrics.increment_streams(1);
+
+ // Dual-write: also update SharedMetadata for consistent cross-shard reads
+ let _ = self.shared_metadata.create_stream(name);
+
create_stream_file_hierarchy(stream.id(), &self.config.system).await?;
Ok(stream)
}
@@ -100,8 +104,12 @@ impl IggyShard {
self.streams.with_index_mut(|index| {
// Rename the key inside of hashmap
let idx = index.remove(&old_name).expect("Rename key: key not found");
- index.insert(name, idx);
+ index.insert(name.clone(), idx);
});
+
+ // Dual-write: also update SharedMetadata
+ let _ = self.shared_metadata.update_stream(id, name);
+
Ok(())
}
@@ -147,6 +155,9 @@ impl IggyShard {
let mut stream = self.delete_stream_base(id);
let stream_id_usize = stream.id();
+ // Dual-write: also delete from SharedMetadata
+ let _ = self.shared_metadata.delete_stream(id);
+
// Clean up consumer groups from ClientManager for this stream
self.client_manager
.delete_consumer_groups_for_stream(stream_id_usize);
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 3b1c86063..a5830c8fd 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -72,7 +72,7 @@ impl IggyShard {
let topic = topic::create_and_insert_topics_mem(
&self.streams,
stream_id,
- name,
+ name.clone(),
replication_factor.unwrap_or(1),
message_expiry,
compression,
@@ -81,6 +81,16 @@ impl IggyShard {
);
self.metrics.increment_topics(1);
+ // Dual-write: also update SharedMetadata
+ let _ = self.shared_metadata.create_topic(
+ stream_id,
+ name,
+ replication_factor.unwrap_or(1),
+ message_expiry,
+ compression,
+ max_topic_size,
+ );
+
// Create file hierarchy for the topic.
create_topic_file_hierarchy(numeric_stream_id, topic.id(), &self.config.system).await?;
Ok(topic)
@@ -191,9 +201,20 @@ impl IggyShard {
self.streams
.with_topic_by_id_mut(stream_id, topic_id, update_topic_closure);
if old_name != new_name {
- let rename_closure = topics::helpers::rename_index(&old_name, new_name);
+ let rename_closure = topics::helpers::rename_index(&old_name, new_name.clone());
self.streams.with_topics(stream_id, rename_closure);
}
+
+ // Dual-write: also update SharedMetadata
+ let _ = self.shared_metadata.update_topic(
+ stream_id,
+ topic_id,
+ Some(name),
+ Some(message_expiry),
+ Some(compression_algorithm),
+ Some(max_topic_size),
+ Some(replication_factor),
+ );
}
pub async fn delete_topic(
@@ -222,6 +243,9 @@ impl IggyShard {
let mut topic = self.delete_topic_base(stream_id, topic_id);
let topic_id_numeric = topic.id();
+ // Dual-write: also delete from SharedMetadata
+ let _ = self.shared_metadata.delete_topic(stream_id, topic_id);
+
// Clean up consumer groups from ClientManager for this topic
self.client_manager
.delete_consumer_groups_for_topic(numeric_stream_id, topic_id_numeric);
(iggy) 04/04: XD
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch arc-swap
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c7fe0e2c87cf20bbcdff09599f4da202872f0f0f
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Dec 29 16:36:01 2025 +0100
XD
---
core/configs/server.toml | 2 +-
core/server/src/http/http_server.rs | 34 +--
core/server/src/main.rs | 136 ++++++++-
core/server/src/metadata/mod.rs | 2 +
core/server/src/metadata/shared.rs | 62 +++-
core/server/src/metadata/stats_store.rs | 108 +++++++
core/server/src/quic/quic_server.rs | 14 +-
core/server/src/shard/builder.rs | 10 +-
core/server/src/shard/handlers.rs | 50 ++--
core/server/src/shard/lazy_init.rs | 317 +++++++++++++++++++++
core/server/src/shard/mod.rs | 6 +-
core/server/src/shard/system/partitions.rs | 9 +
core/server/src/shard/system/streams.rs | 8 +-
core/server/src/shard/system/topics.rs | 14 +-
.../src/shard/tasks/periodic/message_saver.rs | 17 ++
core/server/src/shard/transmission/event.rs | 7 +-
core/server/src/slab/partitions.rs | 104 ++++---
core/server/src/slab/streams.rs | 99 +++++--
core/server/src/slab/topics.rs | 84 ++++--
core/server/src/streaming/partitions/partition.rs | 2 +-
core/server/src/streaming/topics/topic.rs | 2 +-
core/server/src/tcp/tcp_listener.rs | 14 +-
core/server/src/tcp/tcp_tls_listener.rs | 13 +-
core/server/src/websocket/websocket_listener.rs | 26 +-
.../server/src/websocket/websocket_tls_listener.rs | 26 +-
25 files changed, 952 insertions(+), 214 deletions(-)
diff --git a/core/configs/server.toml b/core/configs/server.toml
index cec71ddc7..c2a521b87 100644
--- a/core/configs/server.toml
+++ b/core/configs/server.toml
@@ -152,7 +152,7 @@ enabled = true
address = "127.0.0.1:8090"
# Enable TCP socket migration across shards.
-socket_migration = true
+socket_migration = false
# Whether to use ipv4 or ipv6
ipv6 = false
diff --git a/core/server/src/http/http_server.rs b/core/server/src/http/http_server.rs
index 23eefbdab..6abc5fbc7 100644
--- a/core/server/src/http/http_server.rs
+++ b/core/server/src/http/http_server.rs
@@ -28,7 +28,6 @@ use crate::http::*;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::tasks::periodic::spawn_jwt_token_cleaner;
-use crate::shard::transmission::event::ShardEvent;
use crate::streaming::persistence::persister::PersisterKind;
use axum::extract::DefaultBodyLimit;
use axum::extract::connect_info::Connected;
@@ -38,7 +37,6 @@ use axum_server::tls_rustls::RustlsConfig;
use compio_net::TcpListener;
use err_trail::ErrContext;
use iggy_common::IggyError;
-use iggy_common::TransportProtocol;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::rc::Rc;
@@ -142,15 +140,12 @@ pub async fn start_http_server(
.expect("Failed to get local address for HTTP server");
info!("Started {api_name} on: {address}");
- // Notify shard about the bound address
- let event = ShardEvent::AddressBound {
- protocol: TransportProtocol::Http,
- address,
- };
-
- crate::shard::handlers::handle_event(&shard, event)
- .await
- .ok();
+ // Store in Cell for config_writer backward compat
+ shard.http_bound_address.set(Some(address));
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_http_address(address);
+ // Notify config_writer that HTTP is bound
+ let _ = shard.config_writer_notify.try_send(());
let service = app.into_make_service_with_connect_info::<CompioSocketAddr>();
@@ -187,17 +182,12 @@ pub async fn start_http_server(
info!("Started {api_name} on: {address}");
- // Notify shard about the bound address
- use crate::shard::transmission::event::ShardEvent;
- use iggy_common::TransportProtocol;
- let event = ShardEvent::AddressBound {
- protocol: TransportProtocol::Http,
- address,
- };
-
- crate::shard::handlers::handle_event(&shard, event)
- .await
- .ok();
+ // Store in Cell for config_writer backward compat
+ shard.http_bound_address.set(Some(address));
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_http_address(address);
+ // Notify config_writer that HTTP is bound
+ let _ = shard.config_writer_notify.try_send(());
let service = app.into_make_service_with_connect_info::<SocketAddr>();
let handle = axum_server::Handle::new();
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 51b7f0566..9e63dbc12 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -34,15 +34,21 @@ use server::configs::sharding::ShardAllocator;
use server::diagnostics::{print_io_uring_permission_info, print_locked_memory_limit_info};
use server::io::fs_utils;
use server::log::logger::Logging;
-use server::metadata::SharedMetadata;
+use server::metadata::{
+ ConsumerGroupMeta, MetadataSnapshot, PartitionMeta, SharedMetadata, SharedStatsStore,
+ StreamMeta, TopicMeta, UserMeta,
+};
use server::server_error::ServerError;
use server::shard::namespace::IggyNamespace;
use server::shard::system::info::SystemInfo;
use server::shard::transmission::id::ShardId;
use server::shard::{IggyShard, calculate_shard_assignment};
+use server::slab::Keyed;
+use server::slab::streams::Streams;
use server::slab::traits_ext::{
EntityComponentSystem, EntityComponentSystemMutCell, IntoComponents,
};
+use server::slab::users::Users;
use server::state::file::FileState;
use server::state::system::SystemState;
use server::streaming::clients::client_manager::{Client, ClientManager};
@@ -327,6 +333,14 @@ fn main() -> Result<(), ServerError> {
let shared_metadata = Box::leak(shared_metadata);
let shared_metadata: EternalPtr<SharedMetadata> = shared_metadata.into();
+ // Create shared stats store for cross-shard stats visibility
+ let shared_stats = Box::new(SharedStatsStore::new());
+ let shared_stats = Box::leak(shared_stats);
+ let shared_stats: EternalPtr<SharedStatsStore> = shared_stats.into();
+
+ // Populate SharedMetadata from loaded state (also registers stats in SharedStatsStore)
+ populate_shared_metadata(&shared_metadata, &shared_stats, &streams, &users);
+
streams.with_components(|components| {
let (root, ..) = components.into_components();
for (_, stream) in root.iter() {
@@ -360,6 +374,7 @@ fn main() -> Result<(), ServerError> {
let streams = streams.clone();
let shards_table = shards_table.clone();
let shared_metadata = shared_metadata.clone();
+ let shared_stats = shared_stats.clone();
let users = users.clone();
let connections = connections.clone();
let config = config.clone();
@@ -425,6 +440,7 @@ fn main() -> Result<(), ServerError> {
.users(users)
.shards_table(shards_table)
.shared_metadata(shared_metadata)
+ .shared_stats(shared_stats)
.connections(connections)
.clients_manager(client_manager)
.config(config)
@@ -552,3 +568,121 @@ fn main() -> Result<(), ServerError> {
Ok(())
})
}
+
+/// Populate SharedMetadata with existing streams and users loaded from state.
+/// Also registers stats in SharedStatsStore for cross-shard visibility.
+fn populate_shared_metadata(
+ shared_metadata: &SharedMetadata,
+ shared_stats: &SharedStatsStore,
+ streams: &Streams,
+ users: &Users,
+) {
+ let mut snapshot = MetadataSnapshot::new();
+ let mut max_stream_id: usize = 0;
+ let mut max_user_id: u32 = 0;
+
+ // Populate streams, topics, partitions, consumer groups
+ streams.with_components(|components| {
+ let (root, stats) = components.into_components();
+ for (idx, stream) in root.iter() {
+ max_stream_id = max_stream_id.max(stream.id());
+
+ // Register stream stats in SharedStatsStore
+ let stream_stats = Arc::clone(&stats[idx]);
+ shared_stats.register_stream_stats(stream.id(), stream_stats);
+
+ let mut stream_meta =
+ StreamMeta::new(stream.id(), stream.name().to_string(), stream.created_at());
+
+ stream.topics().with_components(|topic_components| {
+ let (topic_roots, _topic_aux, topic_stats) = topic_components.into_components();
+ for (topic_idx, topic) in topic_roots.iter() {
+ // Register topic stats in SharedStatsStore
+ let topic_stats_arc = Arc::clone(&topic_stats[topic_idx]);
+ shared_stats.register_topic_stats(stream.id(), topic.id(), topic_stats_arc);
+
+ let mut topic_meta = TopicMeta::new(
+ topic.id(),
+ topic.name().to_string(),
+ topic.created_at(),
+ topic.replication_factor(),
+ topic.message_expiry(),
+ topic.compression_algorithm(),
+ topic.max_topic_size(),
+ );
+
+ // Add partitions
+ topic.partitions().with_components(|part_components| {
+ let (part_roots, part_stats, ..) = part_components.into_components();
+ for (part_idx, partition) in part_roots.iter() {
+ // Register partition stats in SharedStatsStore
+ let part_stats_arc = Arc::clone(&part_stats[part_idx]);
+ shared_stats.register_partition_stats(
+ stream.id(),
+ topic.id(),
+ partition.id(),
+ part_stats_arc,
+ );
+
+ topic_meta.add_partition(PartitionMeta::new(
+ partition.id(),
+ partition.created_at(),
+ ));
+ }
+ });
+
+ // Add consumer groups
+ topic.consumer_groups().with_components(|cg_components| {
+ let (cg_roots, ..) = cg_components.into_components();
+ for (_, cg) in cg_roots.iter() {
+ let partition_ids = cg.partitions().clone();
+ topic_meta.add_consumer_group(ConsumerGroupMeta::new(
+ cg.id(),
+ cg.key().clone(),
+ partition_ids,
+ ));
+ }
+ });
+
+ stream_meta.add_topic(topic_meta);
+ }
+ });
+
+ let name = stream_meta.name.clone();
+ snapshot.streams.insert(stream.id(), stream_meta);
+ snapshot.stream_index.insert(name, stream.id());
+ }
+ });
+
+ // Populate users
+ for user in users.values() {
+ max_user_id = max_user_id.max(user.id);
+
+ let user_meta = UserMeta::new(
+ user.id,
+ user.username.clone(),
+ user.password.clone(),
+ user.created_at,
+ user.status,
+ user.permissions.clone(),
+ );
+
+ let username = user_meta.username.clone();
+ snapshot.users.insert(user.id, user_meta);
+ snapshot.user_index.insert(username, user.id);
+ }
+
+ // Capture counts before consuming snapshot
+ let streams_count = snapshot.streams.len();
+ let users_count = snapshot.users.len();
+ let next_stream = max_stream_id.saturating_add(1);
+ let next_user = max_user_id.saturating_add(1);
+
+ // Initialize with next IDs = max + 1
+ shared_metadata.init_from_snapshot(snapshot, next_stream, next_user);
+
+ info!(
+ "Populated SharedMetadata: {} streams, {} users (next_stream_id={}, next_user_id={})",
+ streams_count, users_count, next_stream, next_user,
+ );
+}
diff --git a/core/server/src/metadata/mod.rs b/core/server/src/metadata/mod.rs
index e964dccd9..7564c1e9c 100644
--- a/core/server/src/metadata/mod.rs
+++ b/core/server/src/metadata/mod.rs
@@ -25,6 +25,7 @@ mod consumer_group;
mod partition;
mod shared;
mod snapshot;
+mod stats_store;
mod stream;
mod topic;
mod user;
@@ -33,6 +34,7 @@ pub use consumer_group::ConsumerGroupMeta;
pub use partition::PartitionMeta;
pub use shared::SharedMetadata;
pub use snapshot::MetadataSnapshot;
+pub use stats_store::SharedStatsStore;
pub use stream::StreamMeta;
pub use topic::TopicMeta;
pub use user::{PersonalAccessTokenMeta, UserMeta};
diff --git a/core/server/src/metadata/shared.rs b/core/server/src/metadata/shared.rs
index f03086b65..9e108c94c 100644
--- a/core/server/src/metadata/shared.rs
+++ b/core/server/src/metadata/shared.rs
@@ -131,16 +131,21 @@ impl SharedMetadata {
// Stream operations
- /// Create a new stream. Returns the created stream metadata.
- pub fn create_stream(&self, name: String) -> Result<StreamMeta, IggyError> {
+ /// Add a stream with a specific ID (used by dual-write from shard 0).
+ /// The ID should come from the actual slab allocation.
+ pub fn add_stream(
+ &self,
+ id: usize,
+ name: String,
+ created_at: IggyTimestamp,
+ ) -> Result<StreamMeta, IggyError> {
let current = self.inner.load();
if current.stream_exists_by_name(&name) {
return Err(IggyError::StreamNameAlreadyExists(name));
}
- let id = self.next_stream_id.fetch_add(1, Ordering::SeqCst);
- let stream = StreamMeta::new(id, name.clone(), IggyTimestamp::now());
+ let stream = StreamMeta::new(id, name.clone(), created_at);
let mut new_snapshot = (**current).clone();
new_snapshot.stream_index.insert(name, id);
@@ -151,6 +156,13 @@ impl SharedMetadata {
Ok(stream)
}
+ /// Create a new stream with auto-generated ID.
+ /// Primarily for tests - production code should use `add_stream` with the actual slab ID.
+ pub fn create_stream(&self, name: String) -> Result<StreamMeta, IggyError> {
+ let id = self.next_stream_id.fetch_add(1, Ordering::SeqCst);
+ self.add_stream(id, name, IggyTimestamp::now())
+ }
+
/// Delete a stream. Returns the deleted stream metadata.
pub fn delete_stream(&self, id: &Identifier) -> Result<StreamMeta, IggyError> {
let current = self.inner.load();
@@ -207,12 +219,15 @@ impl SharedMetadata {
// Topic operations
- /// Create a new topic in a stream.
+ /// Add a topic with a specific ID (used by dual-write from shard 0).
+ /// The ID should come from the actual slab allocation.
#[allow(clippy::too_many_arguments)]
- pub fn create_topic(
+ pub fn add_topic(
&self,
stream_id: &Identifier,
+ topic_id: usize,
name: String,
+ created_at: IggyTimestamp,
replication_factor: u8,
message_expiry: IggyExpiry,
compression_algorithm: CompressionAlgorithm,
@@ -229,12 +244,10 @@ impl SharedMetadata {
return Err(IggyError::TopicNameAlreadyExists(name, stream_id.clone()));
}
- // Topic IDs are sequential within a stream
- let topic_id = stream.topics.len();
let topic = TopicMeta::new(
topic_id,
name.clone(),
- IggyTimestamp::now(),
+ created_at,
replication_factor,
message_expiry,
compression_algorithm,
@@ -253,6 +266,37 @@ impl SharedMetadata {
Ok(topic)
}
+ /// Create a new topic with auto-generated ID.
+ /// Primarily for tests - production code should use `add_topic` with the actual slab ID.
+ #[allow(clippy::too_many_arguments)]
+ pub fn create_topic(
+ &self,
+ stream_id: &Identifier,
+ name: String,
+ replication_factor: u8,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ ) -> Result<TopicMeta, IggyError> {
+ let current = self.inner.load();
+ let stream_idx = current
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+ let stream = current.streams.get(&stream_idx).unwrap();
+ let topic_id = stream.topics.len();
+
+ self.add_topic(
+ stream_id,
+ topic_id,
+ name,
+ IggyTimestamp::now(),
+ replication_factor,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ )
+ }
+
/// Delete a topic from a stream.
pub fn delete_topic(
&self,
diff --git a/core/server/src/metadata/stats_store.rs b/core/server/src/metadata/stats_store.rs
new file mode 100644
index 000000000..242e80bca
--- /dev/null
+++ b/core/server/src/metadata/stats_store.rs
@@ -0,0 +1,108 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Shared statistics store for cross-shard stats visibility.
+//!
+//! When shard 0 creates a stream/topic/partition, it registers the stats Arc here.
+//! When other shards do lazy init, they retrieve the same Arc so updates are visible.
+
+use crate::streaming::stats::{PartitionStats, StreamStats, TopicStats};
+use dashmap::DashMap;
+use std::sync::Arc;
+
+/// Thread-safe store for sharing stats Arcs across shards.
+/// Uses DashMap for lock-free concurrent access.
+#[derive(Debug, Default)]
+pub struct SharedStatsStore {
+ /// Stream stats indexed by stream_id
+ stream_stats: DashMap<usize, Arc<StreamStats>>,
+
+ /// Topic stats indexed by (stream_id, topic_id)
+ topic_stats: DashMap<(usize, usize), Arc<TopicStats>>,
+
+ /// Partition stats indexed by (stream_id, topic_id, partition_id)
+ partition_stats: DashMap<(usize, usize, usize), Arc<PartitionStats>>,
+}
+
+impl SharedStatsStore {
+ pub fn new() -> Self {
+ Self {
+ stream_stats: DashMap::new(),
+ topic_stats: DashMap::new(),
+ partition_stats: DashMap::new(),
+ }
+ }
+
+ // Stream stats
+
+ pub fn register_stream_stats(&self, stream_id: usize, stats: Arc<StreamStats>) {
+ self.stream_stats.insert(stream_id, stats);
+ }
+
+ pub fn get_stream_stats(&self, stream_id: usize) -> Option<Arc<StreamStats>> {
+ self.stream_stats.get(&stream_id).map(|r| Arc::clone(&r))
+ }
+
+ pub fn remove_stream_stats(&self, stream_id: usize) {
+ self.stream_stats.remove(&stream_id);
+ }
+
+ // Topic stats
+
+ pub fn register_topic_stats(&self, stream_id: usize, topic_id: usize, stats: Arc<TopicStats>) {
+ self.topic_stats.insert((stream_id, topic_id), stats);
+ }
+
+ pub fn get_topic_stats(&self, stream_id: usize, topic_id: usize) -> Option<Arc<TopicStats>> {
+ self.topic_stats
+ .get(&(stream_id, topic_id))
+ .map(|r| Arc::clone(&r))
+ }
+
+ pub fn remove_topic_stats(&self, stream_id: usize, topic_id: usize) {
+ self.topic_stats.remove(&(stream_id, topic_id));
+ }
+
+ // Partition stats
+
+ pub fn register_partition_stats(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ stats: Arc<PartitionStats>,
+ ) {
+ self.partition_stats
+ .insert((stream_id, topic_id, partition_id), stats);
+ }
+
+ pub fn get_partition_stats(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ ) -> Option<Arc<PartitionStats>> {
+ self.partition_stats
+ .get(&(stream_id, topic_id, partition_id))
+ .map(|r| Arc::clone(&r))
+ }
+
+ pub fn remove_partition_stats(&self, stream_id: usize, topic_id: usize, partition_id: usize) {
+ self.partition_stats
+ .remove(&(stream_id, topic_id, partition_id));
+ }
+}
diff --git a/core/server/src/quic/quic_server.rs b/core/server/src/quic/quic_server.rs
index d70a30877..e18b5a1c9 100644
--- a/core/server/src/quic/quic_server.rs
+++ b/core/server/src/quic/quic_server.rs
@@ -21,7 +21,6 @@ use crate::quic::{COMPONENT, listener, quic_socket};
use crate::server_error::QuicError;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
-use crate::shard::transmission::event::ShardEvent;
use anyhow::Result;
use compio_quic::{
Endpoint, EndpointConfig, IdleTimeout, ServerBuilder, ServerConfig, TransportConfig, VarInt,
@@ -66,7 +65,7 @@ pub async fn spawn_quic_server(
if shard.id != 0 && addr.port() == 0 {
info!("Waiting for QUIC address from shard 0...");
loop {
- if let Some(bound_addr) = shard.quic_bound_address.get() {
+ if let Some(bound_addr) = shard.shared_metadata.load().bound_addresses.quic {
addr = bound_addr;
info!("Received QUIC address: {}", addr);
break;
@@ -117,19 +116,14 @@ pub async fn spawn_quic_server(
info!("Iggy QUIC server has started on: {:?}", actual_addr);
if shard.id == 0 {
- // Store bound address locally
+ // Store in Cell for config_writer backward compat
shard.quic_bound_address.set(Some(actual_addr));
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_quic_address(actual_addr);
if addr.port() == 0 {
// Notify config writer on shard 0
let _ = shard.config_writer_notify.try_send(());
-
- // Broadcast to other shards for SO_REUSEPORT binding
- let event = ShardEvent::AddressBound {
- protocol: iggy_common::TransportProtocol::Quic,
- address: actual_addr,
- };
- shard.broadcast_event_to_all_shards(event).await?;
}
} else {
shard.quic_bound_address.set(Some(actual_addr));
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index e394b9e0c..0996a88b9 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -22,7 +22,7 @@ use super::{
};
use crate::{
configs::server::ServerConfig,
- metadata::SharedMetadata,
+ metadata::{SharedMetadata, SharedStatsStore},
shard::namespace::IggyNamespace,
slab::{streams::Streams, users::Users},
state::file::FileState,
@@ -42,6 +42,7 @@ pub struct IggyShardBuilder {
streams: Option<Streams>,
shards_table: Option<EternalPtr<DashMap<IggyNamespace, ShardId>>>,
shared_metadata: Option<EternalPtr<SharedMetadata>>,
+ shared_stats: Option<EternalPtr<SharedStatsStore>>,
state: Option<FileState>,
users: Option<Users>,
client_manager: Option<ClientManager>,
@@ -82,6 +83,11 @@ impl IggyShardBuilder {
self
}
+ pub fn shared_stats(mut self, shared_stats: EternalPtr<SharedStatsStore>) -> Self {
+ self.shared_stats = Some(shared_stats);
+ self
+ }
+
pub fn clients_manager(mut self, client_manager: ClientManager) -> Self {
self.client_manager = Some(client_manager);
self
@@ -128,6 +134,7 @@ impl IggyShardBuilder {
let streams = self.streams.unwrap();
let shards_table = self.shards_table.unwrap();
let shared_metadata = self.shared_metadata.unwrap();
+ let shared_stats = self.shared_stats.unwrap();
let state = self.state.unwrap();
let users = self.users.unwrap();
let config = self.config.unwrap();
@@ -163,6 +170,7 @@ impl IggyShardBuilder {
shards,
shards_table,
shared_metadata,
+ shared_stats,
streams, // TODO: Fixme
users,
fs_locks: Default::default(),
diff --git a/core/server/src/shard/handlers.rs b/core/server/src/shard/handlers.rs
index c1aed5c92..23c0ea52a 100644
--- a/core/server/src/shard/handlers.rs
+++ b/core/server/src/shard/handlers.rs
@@ -63,6 +63,11 @@ async fn handle_request(
let partition_id = request.partition_id;
match request.payload {
ShardRequestPayload::SendMessages { batch } => {
+ // Lazy init: ensure partition exists locally
+ shard
+ .ensure_local_partition(&stream_id, &topic_id, partition_id)
+ .await?;
+
let ns = IggyFullNamespace::new(stream_id, topic_id, partition_id);
let batch = shard.maybe_encrypt_messages(batch)?;
let messages_count = batch.count();
@@ -74,6 +79,11 @@ async fn handle_request(
Ok(ShardResponse::SendMessages)
}
ShardRequestPayload::PollMessages { args, consumer } => {
+ // Lazy init: ensure partition exists locally
+ shard
+ .ensure_local_partition(&stream_id, &topic_id, partition_id)
+ .await?;
+
let auto_commit = args.auto_commit;
let ns = IggyFullNamespace::new(stream_id, topic_id, partition_id);
let (metadata, batches) = shard.streams.poll_messages(&ns, consumer, args).await?;
@@ -97,12 +107,22 @@ async fn handle_request(
Ok(ShardResponse::PollMessages((metadata, batches)))
}
ShardRequestPayload::FlushUnsavedBuffer { fsync } => {
+ // Lazy init: ensure partition exists locally
+ shard
+ .ensure_local_partition(&stream_id, &topic_id, partition_id)
+ .await?;
+
shard
.flush_unsaved_buffer_base(&stream_id, &topic_id, partition_id, fsync)
.await?;
Ok(ShardResponse::FlushUnsavedBuffer)
}
ShardRequestPayload::DeleteSegments { segments_count } => {
+ // Lazy init: ensure partition exists locally
+ shard
+ .ensure_local_partition(&stream_id, &topic_id, partition_id)
+ .await?;
+
shard
.delete_segments_base(&stream_id, &topic_id, partition_id, segments_count)
.await?;
@@ -296,6 +316,11 @@ async fn handle_request(
let registry = shard.task_registry.clone();
let registry_clone = registry.clone();
+ // Lazy init: ensure partition exists locally
+ shard
+ .ensure_local_partition(&stream_id, &topic_id, partition_id)
+ .await?;
+
let ns = IggyFullNamespace::new(stream_id, topic_id, partition_id);
let batch = shard.maybe_encrypt_messages(initial_data)?;
let messages_count = batch.count();
@@ -358,30 +383,5 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
.await?;
Ok(())
}
- ShardEvent::AddressBound { protocol, address } => {
- info!(
- "Received AddressBound event for {:?} with address: {}",
- protocol, address
- );
- match protocol {
- TransportProtocol::Tcp => {
- shard.tcp_bound_address.set(Some(address));
- let _ = shard.config_writer_notify.try_send(());
- }
- TransportProtocol::Quic => {
- shard.quic_bound_address.set(Some(address));
- let _ = shard.config_writer_notify.try_send(());
- }
- TransportProtocol::Http => {
- shard.http_bound_address.set(Some(address));
- let _ = shard.config_writer_notify.try_send(());
- }
- TransportProtocol::WebSocket => {
- shard.websocket_bound_address.set(Some(address));
- let _ = shard.config_writer_notify.try_send(());
- }
- }
- Ok(())
- }
}
}
diff --git a/core/server/src/shard/lazy_init.rs b/core/server/src/shard/lazy_init.rs
new file mode 100644
index 000000000..06f63b796
--- /dev/null
+++ b/core/server/src/shard/lazy_init.rs
@@ -0,0 +1,317 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Lazy initialization of Stream/Topic/Partition structures.
+//!
+//! When a non-shard-0 receives a data plane request (SendMessages, PollMessages),
+//! it may not have the Stream/Topic/Partition in its local slab. This module
+//! provides lazy creation from SharedMetadata.
+
+use crate::metadata::{PartitionMeta, StreamMeta, TopicMeta};
+use crate::shard::IggyShard;
+use crate::slab::traits_ext::{EntityMarker, Insert, InsertCell};
+use crate::streaming::partitions::helpers::create_message_deduplicator;
+use crate::streaming::partitions::log::SegmentedLog;
+use crate::streaming::partitions::partition::{ConsumerGroupOffsets, ConsumerOffsets, Partition};
+use crate::streaming::partitions::storage::create_partition_file_hierarchy;
+use crate::streaming::stats::{StreamStats, TopicStats};
+use crate::streaming::streams::stream::Stream;
+use crate::streaming::topics::topic::Topic;
+use iggy_common::{Identifier, IggyError};
+use std::sync::Arc;
+use std::sync::atomic::AtomicU64;
+use tracing::{debug, info};
+
+impl IggyShard {
+ /// Ensures the stream/topic/partition exists locally for data operations.
+ /// Creates them from SharedMetadata if they don't exist.
+ ///
+ /// This is the core of lazy initialization - non-shard-0 shards create
+ /// data structures on-demand when they receive data plane requests.
+ ///
+ /// Accepts `Identifier` references (can be numeric or string) and resolves
+ /// them to numeric IDs using SharedMetadata.
+ pub async fn ensure_local_partition(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition_id: usize,
+ ) -> Result<(), IggyError> {
+ // Load metadata to resolve identifiers
+ let metadata = self.shared_metadata.load();
+
+ // Resolve stream identifier to numeric ID
+ let numeric_stream_id = metadata
+ .get_stream_id(stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ // Get stream metadata and resolve topic identifier
+ let stream_meta = metadata
+ .streams
+ .get(&numeric_stream_id)
+ .ok_or_else(|| IggyError::StreamIdNotFound(stream_id.clone()))?;
+
+ let numeric_topic_id = Self::resolve_topic_id(topic_id, stream_meta)
+ .ok_or_else(|| IggyError::TopicIdNotFound(topic_id.clone(), stream_id.clone()))?;
+
+ // Create numeric identifiers for slab lookups
+ let stream_ident = Identifier::numeric(numeric_stream_id as u32)?;
+ let topic_ident = Identifier::numeric(numeric_topic_id as u32)?;
+
+ // Fast path: check if partition already exists
+ if self.streams.exists(&stream_ident) {
+ let topic_exists = self
+ .streams
+ .with_topics(&stream_ident, |topics| topics.exists(&topic_ident));
+
+ if topic_exists {
+ let partition_exists =
+ self.streams
+ .with_partitions(&stream_ident, &topic_ident, |partitions| {
+ partitions.exists(partition_id)
+ });
+
+ if partition_exists {
+ return Ok(());
+ }
+ }
+ }
+
+ // Slow path: need to create from metadata
+ debug!(
+ "Lazy init: stream={}, topic={}, partition={} on shard {}",
+ numeric_stream_id, numeric_topic_id, partition_id, self.id
+ );
+
+ let topic_meta = stream_meta
+ .topics
+ .get(&numeric_topic_id)
+ .ok_or_else(|| IggyError::TopicIdNotFound(topic_id.clone(), stream_id.clone()))?;
+
+ let partition_meta =
+ topic_meta
+ .partitions
+ .get(&partition_id)
+ .ok_or(IggyError::PartitionNotFound(
+ partition_id,
+ topic_ident.clone(),
+ stream_ident.clone(),
+ ))?;
+
+ // Create stream if needed (and get its stats)
+ let stream_stats = if !self.streams.exists(&stream_ident) {
+ self.create_stream_from_meta(stream_meta)?
+ } else {
+ // Get existing stream's stats
+ self.streams
+ .with_stream_by_id(&stream_ident, |(_root, stats)| Arc::clone(&stats))
+ };
+
+ // Create topic if needed (and get its stats)
+ let topic_exists = self
+ .streams
+ .with_topics(&stream_ident, |topics| topics.exists(&topic_ident));
+
+ let topic_stats = if !topic_exists {
+ self.create_topic_from_meta(numeric_stream_id, topic_meta, stream_stats)?
+ } else {
+ // Get existing topic's stats
+ self.streams
+ .with_topic_by_id(&stream_ident, &topic_ident, |(_root, _, stats)| {
+ Arc::clone(&stats)
+ })
+ };
+
+ // Create partition if needed
+ let partition_exists =
+ self.streams
+ .with_partitions(&stream_ident, &topic_ident, |partitions| {
+ partitions.exists(partition_id)
+ });
+
+ if !partition_exists {
+ self.create_partition_from_meta(
+ numeric_stream_id,
+ numeric_topic_id,
+ partition_meta,
+ topic_stats,
+ )
+ .await?;
+ }
+
+ Ok(())
+ }
+
+ /// Resolves a topic identifier to its numeric ID.
+ fn resolve_topic_id(topic_id: &Identifier, stream_meta: &StreamMeta) -> Option<usize> {
+ match topic_id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = topic_id.get_u32_value().unwrap() as usize;
+ if stream_meta.topics.contains_key(&id) {
+ Some(id)
+ } else {
+ None
+ }
+ }
+ iggy_common::IdKind::String => {
+ let name = topic_id.get_string_value().unwrap();
+ stream_meta.get_topic_id_by_name(&name)
+ }
+ }
+ }
+
+ /// Creates a Stream locally from metadata. Returns the stream stats.
+ /// Uses SharedStatsStore to get the same Arc<StreamStats> as shard 0.
+ fn create_stream_from_meta(&self, meta: &StreamMeta) -> Result<Arc<StreamStats>, IggyError> {
+ info!(
+ "Lazy creating stream: id={}, name={} on shard {}",
+ meta.id, meta.name, self.id
+ );
+
+ // Get the shared stats from SharedStatsStore (registered by shard 0)
+ let stats = self.shared_stats.get_stream_stats(meta.id).ok_or_else(|| {
+ IggyError::StreamIdNotFound(Identifier::numeric(meta.id as u32).unwrap())
+ })?;
+
+ let mut stream = Stream::new(meta.name.clone(), Arc::clone(&stats), meta.created_at);
+ stream.update_id(meta.id);
+
+ // Insert into local streams slab
+ // Note: slab position may differ from logical ID - that's fine,
+ // lookups use the name index which maps name -> slab position
+ let _ = self.streams.insert(stream);
+
+ Ok(stats)
+ }
+
+ /// Creates a Topic locally from metadata. Returns the topic stats.
+ /// Uses SharedStatsStore to get the same Arc<TopicStats> as shard 0.
+ fn create_topic_from_meta(
+ &self,
+ stream_id: usize,
+ meta: &TopicMeta,
+ _stream_stats: Arc<StreamStats>,
+ ) -> Result<Arc<TopicStats>, IggyError> {
+ info!(
+ "Lazy creating topic: stream={}, id={}, name={} on shard {}",
+ stream_id, meta.id, meta.name, self.id
+ );
+
+ // Get the shared stats from SharedStatsStore (registered by shard 0)
+ let stats = self
+ .shared_stats
+ .get_topic_stats(stream_id, meta.id)
+ .ok_or_else(|| {
+ IggyError::TopicIdNotFound(
+ Identifier::numeric(meta.id as u32).unwrap(),
+ Identifier::numeric(stream_id as u32).unwrap(),
+ )
+ })?;
+
+ let mut topic = Topic::new(
+ meta.name.clone(),
+ Arc::clone(&stats),
+ meta.created_at,
+ meta.replication_factor,
+ meta.message_expiry,
+ meta.compression_algorithm,
+ meta.max_topic_size,
+ );
+ topic.update_id(meta.id);
+
+ // Insert into stream's topics
+ // Note: slab position may differ from logical ID
+ let stream_ident = Identifier::numeric(stream_id as u32)?;
+ self.streams.with_topics(&stream_ident, |topics| {
+ let _ = topics.insert(topic);
+ });
+
+ Ok(stats)
+ }
+
+ /// Creates a Partition locally from metadata with SegmentedLog.
+ /// Uses SharedStatsStore to get the same Arc<PartitionStats> as shard 0.
+ async fn create_partition_from_meta(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_meta: &PartitionMeta,
+ _topic_stats: Arc<TopicStats>,
+ ) -> Result<(), IggyError> {
+ let partition_id = partition_meta.id;
+
+ info!(
+ "Lazy creating partition: stream={}, topic={}, partition={} on shard {}",
+ stream_id, topic_id, partition_id, self.id
+ );
+
+ // Create partition directory if needed
+ create_partition_file_hierarchy(stream_id, topic_id, partition_id, &self.config.system)
+ .await?;
+
+ // Get the shared stats from SharedStatsStore (registered by shard 0)
+ let stats = self
+ .shared_stats
+ .get_partition_stats(stream_id, topic_id, partition_id)
+ .ok_or_else(|| {
+ IggyError::PartitionNotFound(
+ partition_id,
+ Identifier::numeric(topic_id as u32).unwrap(),
+ Identifier::numeric(stream_id as u32).unwrap(),
+ )
+ })?;
+
+ // Create message deduplicator if configured
+ let message_deduplicator = create_message_deduplicator(&self.config.system);
+
+ // Create empty SegmentedLog
+ let log = SegmentedLog::default();
+
+ // Create partition
+ let partition = Partition::new(
+ partition_meta.created_at,
+ true, // should_increment_offset
+ stats,
+ message_deduplicator,
+ Arc::new(AtomicU64::new(0)),
+ Arc::new(ConsumerOffsets::with_capacity(10)),
+ Arc::new(ConsumerGroupOffsets::with_capacity(10)),
+ log,
+ );
+
+ // Insert partition into topic
+ let stream_ident = Identifier::numeric(stream_id as u32)?;
+ let topic_ident = Identifier::numeric(topic_id as u32)?;
+
+ // Note: slab position may differ from logical ID
+ self.streams
+ .with_partitions_mut(&stream_ident, &topic_ident, |partitions| {
+ let _ = partitions.insert(partition);
+ });
+
+ // Initialize the log with a segment
+ self.init_log(&stream_ident, &topic_ident, partition_id)
+ .await?;
+
+ info!(
+ "Lazy created partition: stream={}, topic={}, partition={} on shard {}",
+ stream_id, topic_id, partition_id, self.id
+ );
+
+ Ok(())
+ }
+}
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index b5424cd9f..bb2e0266e 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -25,6 +25,7 @@ pub mod transmission;
mod communication;
pub mod handlers;
+mod lazy_init;
// Re-export for backwards compatibility
pub use communication::calculate_shard_assignment;
@@ -33,7 +34,7 @@ use self::tasks::{continuous, periodic};
use crate::{
configs::server::ServerConfig,
io::fs_locks::FsLocks,
- metadata::SharedMetadata,
+ metadata::{SharedMetadata, SharedStatsStore},
shard::{
namespace::IggyNamespace, task_registry::TaskRegistry, transmission::frame::ShardFrame,
},
@@ -77,6 +78,9 @@ pub struct IggyShard {
/// Shared metadata accessible by all shards (ArcSwap-based).
pub(crate) shared_metadata: EternalPtr<SharedMetadata>,
+ /// Shared stats store for cross-shard stats visibility.
+ pub(crate) shared_stats: EternalPtr<SharedStatsStore>,
+
pub(crate) fs_locks: FsLocks,
pub(crate) encryptor: Option<EncryptorKind>,
pub(crate) config: ServerConfig,
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 10177bab3..b65ba443d 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -35,6 +35,7 @@ use crate::streaming::topics;
use err_trail::ErrContext;
use iggy_common::Identifier;
use iggy_common::IggyError;
+use std::sync::Arc;
use tracing::info;
impl IggyShard {
@@ -107,6 +108,14 @@ impl IggyShard {
let shards_count = self.get_available_shards_count();
for (partition_id, stats) in partitions.iter().map(|p| (p.id(), p.stats())) {
+ // Register partition stats in SharedStatsStore for cross-shard visibility
+ self.shared_stats.register_partition_stats(
+ numeric_stream_id,
+ numeric_topic_id,
+ partition_id,
+ Arc::clone(stats),
+ );
+
let ns = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
let shard_id = ShardId::new(calculate_shard_assignment(&ns, shards_count));
let is_current_shard = self.id == *shard_id;
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index b12ca2f2f..4cb9bc041 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -45,8 +45,14 @@ impl IggyShard {
let stream = stream::create_and_insert_stream_mem(&self.streams, name.clone());
self.metrics.increment_streams(1);
+ // Register stats in SharedStatsStore for cross-shard visibility
+ self.shared_stats
+ .register_stream_stats(stream.id(), stream.stats().clone());
+
// Dual-write: also update SharedMetadata for consistent cross-shard reads
- let _ = self.shared_metadata.create_stream(name);
+ let _ = self
+ .shared_metadata
+ .add_stream(stream.id(), name, stream.root().created_at());
create_stream_file_hierarchy(stream.id(), &self.config.system).await?;
Ok(stream)
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index a5830c8fd..83985c279 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -26,6 +26,7 @@ use crate::streaming::{partitions, streams, topics};
use err_trail::ErrContext;
use iggy_common::{CompressionAlgorithm, Identifier, IggyError, IggyExpiry, MaxTopicSize};
use std::str::FromStr;
+use std::sync::Arc;
use tracing::info;
impl IggyShard {
@@ -81,10 +82,19 @@ impl IggyShard {
);
self.metrics.increment_topics(1);
- // Dual-write: also update SharedMetadata
- let _ = self.shared_metadata.create_topic(
+ // Register stats in SharedStatsStore for cross-shard visibility
+ self.shared_stats.register_topic_stats(
+ numeric_stream_id,
+ topic.id(),
+ Arc::clone(topic.stats()),
+ );
+
+ // Dual-write: also update SharedMetadata with actual topic ID from slab
+ let _ = self.shared_metadata.add_topic(
stream_id,
+ topic.id(),
name,
+ topic.root().created_at(),
replication_factor.unwrap_or(1),
message_expiry,
compression,
diff --git a/core/server/src/shard/tasks/periodic/message_saver.rs b/core/server/src/shard/tasks/periodic/message_saver.rs
index c6d0efdcf..25ab6b53c 100644
--- a/core/server/src/shard/tasks/periodic/message_saver.rs
+++ b/core/server/src/shard/tasks/periodic/message_saver.rs
@@ -54,6 +54,15 @@ async fn save_messages(shard: Rc<IggyShard>) -> Result<(), IggyError> {
let topic_id = Identifier::numeric(ns.topic_id() as u32).unwrap();
let partition_id = ns.partition_id();
+ // Skip partitions that haven't been lazily created yet
+ // (they have no messages to save anyway)
+ if !shard
+ .streams
+ .partition_exists(&stream_id, &topic_id, partition_id)
+ {
+ continue;
+ }
+
match shard
.streams
.persist_messages(
@@ -102,6 +111,14 @@ async fn fsync_all_segments_on_shutdown(shard: Rc<IggyShard>, result: Result<(),
let topic_id = Identifier::numeric(ns.topic_id() as u32).unwrap();
let partition_id = ns.partition_id();
+ // Skip partitions that haven't been lazily created yet
+ if !shard
+ .streams
+ .partition_exists(&stream_id, &topic_id, partition_id)
+ {
+ continue;
+ }
+
match shard
.streams
.fsync_all_messages(&stream_id, &topic_id, partition_id)
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 3364a55aa..9c362d9a7 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -15,8 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_common::{Identifier, TransportProtocol};
-use std::net::SocketAddr;
+use iggy_common::Identifier;
use strum::Display;
#[derive(Debug, Clone, Display)]
@@ -28,8 +27,4 @@ pub enum ShardEvent {
partition_id: usize,
fsync: bool,
},
- AddressBound {
- protocol: TransportProtocol,
- address: SocketAddr,
- },
}
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index 0ad5abbf0..a8b990ec6 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -42,6 +42,8 @@ pub type ContainerId = usize;
#[derive(Debug)]
pub struct Partitions {
+ /// Numeric ID index: logical partition ID → slab position
+ numeric_index: ahash::AHashMap<usize, ContainerId>,
root: Slab<partition::PartitionRoot>,
stats: Slab<Arc<PartitionStats>>,
message_deduplicator: Slab<Option<Arc<MessageDeduplicator>>>,
@@ -60,6 +62,7 @@ pub struct Partitions {
impl Clone for Partitions {
fn clone(&self) -> Self {
Self {
+ numeric_index: self.numeric_index.clone(),
root: self.root.clone(),
stats: self.stats.clone(),
message_deduplicator: self.message_deduplicator.clone(),
@@ -79,40 +82,51 @@ impl Insert for Partitions {
let (root, stats, deduplicator, offset, consumer_offset, consumer_group_offset, log) =
item.into_components();
- let entity_id = self.root.insert(root);
- let id = self.stats.insert(stats);
+ // Check if ID was pre-set (e.g., from metadata during lazy creation)
+ let pre_set_id = root.id();
+
+ let slab_pos = self.root.insert(root);
+ let stats_pos = self.stats.insert(stats);
assert_eq!(
- entity_id, id,
- "partition_insert: id mismatch when creating stats"
+ slab_pos, stats_pos,
+ "partition_insert: position mismatch when creating stats"
);
- let id = self.log.insert(log);
+ let log_pos = self.log.insert(log);
assert_eq!(
- entity_id, id,
- "partition_insert: id mismatch when creating log"
+ slab_pos, log_pos,
+ "partition_insert: position mismatch when creating log"
);
- let id = self.message_deduplicator.insert(deduplicator);
+ let dedup_pos = self.message_deduplicator.insert(deduplicator);
assert_eq!(
- entity_id, id,
- "partition_insert: id mismatch when creating message_deduplicator"
+ slab_pos, dedup_pos,
+ "partition_insert: position mismatch when creating message_deduplicator"
);
- let id = self.offset.insert(offset);
+ let offset_pos = self.offset.insert(offset);
assert_eq!(
- entity_id, id,
- "partition_insert: id mismatch when creating offset"
+ slab_pos, offset_pos,
+ "partition_insert: position mismatch when creating offset"
);
- let id = self.consumer_offset.insert(consumer_offset);
+ let consumer_pos = self.consumer_offset.insert(consumer_offset);
assert_eq!(
- entity_id, id,
- "partition_insert: id mismatch when creating consumer_offset"
+ slab_pos, consumer_pos,
+ "partition_insert: position mismatch when creating consumer_offset"
);
- let id = self.consumer_group_offset.insert(consumer_group_offset);
+ let cg_pos = self.consumer_group_offset.insert(consumer_group_offset);
assert_eq!(
- entity_id, id,
- "partition_insert: id mismatch when creating consumer_group_offset"
+ slab_pos, cg_pos,
+ "partition_insert: position mismatch when creating consumer_group_offset"
);
- let root = self.root.get_mut(entity_id).unwrap();
- root.update_id(entity_id);
- entity_id
+
+ let root = self.root.get_mut(slab_pos).unwrap();
+
+ // Use pre-set ID if available, otherwise use slab position
+ let logical_id = if pre_set_id > 0 { pre_set_id } else { slab_pos };
+ root.update_id(logical_id);
+
+ // Update numeric index
+ self.numeric_index.insert(logical_id, slab_pos);
+
+ logical_id
}
}
@@ -120,14 +134,25 @@ impl Delete for Partitions {
type Idx = ContainerId;
type Item = Partition;
- fn delete(&mut self, id: Self::Idx) -> Self::Item {
- let root = self.root.remove(id);
- let stats = self.stats.remove(id);
- let message_deduplicator = self.message_deduplicator.remove(id);
- let offset = self.offset.remove(id);
- let consumer_offset = self.consumer_offset.remove(id);
- let consumer_group_offset = self.consumer_group_offset.remove(id);
- let log = self.log.remove(id);
+ fn delete(&mut self, logical_id: Self::Idx) -> Self::Item {
+ // Look up slab position from logical ID
+ let slab_pos = *self
+ .numeric_index
+ .get(&logical_id)
+ .expect("partition_delete: logical ID not found in numeric index");
+
+ let root = self.root.remove(slab_pos);
+ let stats = self.stats.remove(slab_pos);
+ let message_deduplicator = self.message_deduplicator.remove(slab_pos);
+ let offset = self.offset.remove(slab_pos);
+ let consumer_offset = self.consumer_offset.remove(slab_pos);
+ let consumer_group_offset = self.consumer_group_offset.remove(slab_pos);
+ let log = self.log.remove(slab_pos);
+
+ // Remove from numeric index
+ self.numeric_index
+ .remove(&logical_id)
+ .expect("partition_delete: ID not found in numeric index");
Partition::new_with_components(
root,
@@ -197,6 +222,7 @@ impl EntityComponentSystemMut for Partitions {
impl Default for Partitions {
fn default() -> Self {
Self {
+ numeric_index: ahash::AHashMap::with_capacity(PARTITIONS_CAPACITY),
root: Slab::with_capacity(PARTITIONS_CAPACITY),
stats: Slab::with_capacity(PARTITIONS_CAPACITY),
log: Slab::with_capacity(PARTITIONS_CAPACITY),
@@ -223,21 +249,29 @@ impl Partitions {
pub fn with_partition_by_id<T>(
&self,
- id: ContainerId,
+ logical_id: ContainerId,
f: impl FnOnce(ComponentsById<PartitionRef>) -> T,
) -> T {
- self.with_components_by_id(id, |components| f(components))
+ let slab_pos = *self
+ .numeric_index
+ .get(&logical_id)
+ .expect("Partition not found by ID");
+ self.with_components_by_id(slab_pos, |components| f(components))
}
pub fn exists(&self, id: ContainerId) -> bool {
- self.root.contains(id)
+ self.numeric_index.contains_key(&id)
}
pub fn with_partition_by_id_mut<T>(
&mut self,
- id: ContainerId,
+ logical_id: ContainerId,
f: impl FnOnce(ComponentsById<PartitionRefMut>) -> T,
) -> T {
- self.with_components_by_id_mut(id, |components| f(components))
+ let slab_pos = *self
+ .numeric_index
+ .get(&logical_id)
+ .expect("Partition not found by ID");
+ self.with_components_by_id_mut(slab_pos, |components| f(components))
}
}
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index ad7d5368d..d82b95ebb 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -71,7 +71,11 @@ pub type ContainerId = usize;
#[derive(Debug, Clone)]
pub struct Streams {
+ /// Name index: stream name → slab position
index: RefCell<AHashMap<<stream::StreamRoot as Keyed>::Key, ContainerId>>,
+ /// Numeric ID index: logical stream ID → slab position
+ /// This allows logical IDs to differ from slab positions (needed for lazy creation)
+ numeric_index: RefCell<AHashMap<usize, ContainerId>>,
root: RefCell<Slab<stream::StreamRoot>>,
stats: RefCell<Slab<Arc<StreamStats>>>,
}
@@ -80,6 +84,7 @@ impl Default for Streams {
fn default() -> Self {
Self {
index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
+ numeric_index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
root: RefCell::new(Slab::with_capacity(CAPACITY)),
stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
@@ -109,20 +114,32 @@ impl InsertCell for Streams {
fn insert(&self, item: Self::Item) -> Self::Idx {
let (root, stats) = item.into_components();
let mut root_container = self.root.borrow_mut();
- let mut indexes = self.index.borrow_mut();
+ let mut name_index = self.index.borrow_mut();
+ let mut numeric_index = self.numeric_index.borrow_mut();
let mut stats_container = self.stats.borrow_mut();
+ // Check if ID was pre-set (e.g., from metadata during lazy creation)
+ let pre_set_id = root.id();
+
let key = root.key().clone();
- let entity_id = root_container.insert(root);
- let id = stats_container.insert(stats);
+ let slab_pos = root_container.insert(root);
+ let stats_pos = stats_container.insert(stats);
assert_eq!(
- entity_id, id,
- "stream_insert: id mismatch when inserting stats"
+ slab_pos, stats_pos,
+ "stream_insert: position mismatch when inserting stats"
);
- let root = root_container.get_mut(entity_id).unwrap();
- root.update_id(entity_id);
- indexes.insert(key, entity_id);
- entity_id
+
+ let root = root_container.get_mut(slab_pos).unwrap();
+
+ // Use pre-set ID if available, otherwise use slab position
+ let logical_id = if pre_set_id > 0 { pre_set_id } else { slab_pos };
+ root.update_id(logical_id);
+
+ // Update both indexes
+ name_index.insert(key, slab_pos);
+ numeric_index.insert(logical_id, slab_pos);
+
+ logical_id
}
}
@@ -130,19 +147,28 @@ impl DeleteCell for Streams {
type Idx = ContainerId;
type Item = stream::Stream;
- fn delete(&self, id: Self::Idx) -> Self::Item {
+ fn delete(&self, logical_id: Self::Idx) -> Self::Item {
let mut root_container = self.root.borrow_mut();
- let mut indexes = self.index.borrow_mut();
+ let mut name_index = self.index.borrow_mut();
+ let mut numeric_index = self.numeric_index.borrow_mut();
let mut stats_container = self.stats.borrow_mut();
- let root = root_container.remove(id);
- let stats = stats_container.remove(id);
+ // Look up slab position from logical ID
+ let slab_pos = *numeric_index
+ .get(&logical_id)
+ .expect("stream_delete: logical ID not found in numeric index");
- // Remove from index
+ let root = root_container.remove(slab_pos);
+ let stats = stats_container.remove(slab_pos);
+
+ // Remove from both indexes
let key = root.key();
- indexes
+ name_index
.remove(key)
- .expect("stream_delete: key not found in index");
+ .expect("stream_delete: key not found in name index");
+ numeric_index
+ .remove(&logical_id)
+ .expect("stream_delete: ID not found in numeric index");
stream::Stream::new_with_components(root, stats)
}
@@ -441,8 +467,8 @@ impl Streams {
pub fn exists(&self, id: &Identifier) -> bool {
match id.kind {
iggy_common::IdKind::Numeric => {
- let id = id.get_u32_value().unwrap() as usize;
- self.root.borrow().contains(id)
+ let logical_id = id.get_u32_value().unwrap() as usize;
+ self.numeric_index.borrow().contains_key(&logical_id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
@@ -451,12 +477,45 @@ impl Streams {
}
}
+ /// Check if a partition exists locally (stream, topic, and partition all exist).
+ /// Returns false if any level doesn't exist, without panicking.
+ pub fn partition_exists(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition_id: usize,
+ ) -> bool {
+ if !self.exists(stream_id) {
+ return false;
+ }
+
+ let topic_exists = self.with_topics(stream_id, |topics| topics.exists(topic_id));
+ if !topic_exists {
+ return false;
+ }
+
+ self.with_partitions(stream_id, topic_id, |partitions| {
+ partitions.exists(partition_id)
+ })
+ }
+
pub fn get_index(&self, id: &Identifier) -> usize {
match id.kind {
- iggy_common::IdKind::Numeric => id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::Numeric => {
+ let logical_id = id.get_u32_value().unwrap() as usize;
+ *self
+ .numeric_index
+ .borrow()
+ .get(&logical_id)
+ .expect("Stream not found by numeric ID")
+ }
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
- *self.index.borrow().get(&key).expect("Stream not found")
+ *self
+ .index
+ .borrow()
+ .get(&key)
+ .expect("Stream not found by name")
}
}
}
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 7bfed97e7..39cda6946 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -45,7 +45,10 @@ pub type ContainerId = usize;
#[derive(Debug, Clone)]
pub struct Topics {
+ /// Name index: topic name → slab position
index: RefCell<AHashMap<<topic::TopicRoot as Keyed>::Key, ContainerId>>,
+ /// Numeric ID index: logical topic ID → slab position
+ numeric_index: RefCell<AHashMap<usize, ContainerId>>,
root: RefCell<Slab<topic::TopicRoot>>,
auxilaries: RefCell<Slab<topic::TopicAuxilary>>,
stats: RefCell<Slab<Arc<TopicStats>>>,
@@ -59,25 +62,37 @@ impl InsertCell for Topics {
let (root, auxilary, stats) = item.into_components();
let mut root_container = self.root.borrow_mut();
let mut auxilaries = self.auxilaries.borrow_mut();
- let mut indexes = self.index.borrow_mut();
+ let mut name_index = self.index.borrow_mut();
+ let mut numeric_index = self.numeric_index.borrow_mut();
let mut stats_container = self.stats.borrow_mut();
+ // Check if ID was pre-set (e.g., from metadata during lazy creation)
+ let pre_set_id = root.id();
+
let key = root.key().clone();
- let entity_id = root_container.insert(root);
- let id = stats_container.insert(stats);
+ let slab_pos = root_container.insert(root);
+ let stats_pos = stats_container.insert(stats);
assert_eq!(
- entity_id, id,
- "topic_insert: id mismatch when inserting stats component"
+ slab_pos, stats_pos,
+ "topic_insert: position mismatch when inserting stats component"
);
- let id = auxilaries.insert(auxilary);
+ let aux_pos = auxilaries.insert(auxilary);
assert_eq!(
- entity_id, id,
- "topic_insert: id mismatch when inserting auxilary component"
+ slab_pos, aux_pos,
+ "topic_insert: position mismatch when inserting auxilary component"
);
- let root = root_container.get_mut(entity_id).unwrap();
- root.update_id(entity_id);
- indexes.insert(key, entity_id);
- entity_id
+
+ let root = root_container.get_mut(slab_pos).unwrap();
+
+ // Use pre-set ID if available, otherwise use slab position
+ let logical_id = if pre_set_id > 0 { pre_set_id } else { slab_pos };
+ root.update_id(logical_id);
+
+ // Update both indexes
+ name_index.insert(key, slab_pos);
+ numeric_index.insert(logical_id, slab_pos);
+
+ logical_id
}
}
@@ -85,24 +100,33 @@ impl DeleteCell for Topics {
type Idx = ContainerId;
type Item = topic::Topic;
- fn delete(&self, id: Self::Idx) -> Self::Item {
+ fn delete(&self, logical_id: Self::Idx) -> Self::Item {
let mut root_container = self.root.borrow_mut();
let mut auxilaries = self.auxilaries.borrow_mut();
- let mut indexes = self.index.borrow_mut();
+ let mut name_index = self.index.borrow_mut();
+ let mut numeric_index = self.numeric_index.borrow_mut();
let mut stats_container = self.stats.borrow_mut();
- let root = root_container.remove(id);
- let auxilary = auxilaries.remove(id);
- let stats = stats_container.remove(id);
+ // Look up slab position from logical ID
+ let slab_pos = *numeric_index
+ .get(&logical_id)
+ .expect("topic_delete: logical ID not found in numeric index");
+
+ let root = root_container.remove(slab_pos);
+ let auxilary = auxilaries.remove(slab_pos);
+ let stats = stats_container.remove(slab_pos);
- // Remove from index
+ // Remove from both indexes
let key = root.key();
- indexes.remove(key).unwrap_or_else(|| {
+ name_index.remove(key).unwrap_or_else(|| {
panic!(
"topic_delete: key not found with key: {} and id: {}",
- key, id
+ key, logical_id
)
});
+ numeric_index
+ .remove(&logical_id)
+ .expect("topic_delete: ID not found in numeric index");
topic::Topic::new_with_components(root, auxilary, stats)
}
@@ -130,6 +154,7 @@ impl Default for Topics {
fn default() -> Self {
Self {
index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
+ numeric_index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
root: RefCell::new(Slab::with_capacity(CAPACITY)),
auxilaries: RefCell::new(Slab::with_capacity(CAPACITY)),
stats: RefCell::new(Slab::with_capacity(CAPACITY)),
@@ -173,8 +198,8 @@ impl Topics {
pub fn exists(&self, id: &Identifier) -> bool {
match id.kind {
iggy_common::IdKind::Numeric => {
- let id = id.get_u32_value().unwrap() as usize;
- self.root.borrow().contains(id)
+ let logical_id = id.get_u32_value().unwrap() as usize;
+ self.numeric_index.borrow().contains_key(&logical_id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
@@ -185,10 +210,21 @@ impl Topics {
pub fn get_index(&self, id: &Identifier) -> usize {
match id.kind {
- iggy_common::IdKind::Numeric => id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::Numeric => {
+ let logical_id = id.get_u32_value().unwrap() as usize;
+ *self
+ .numeric_index
+ .borrow()
+ .get(&logical_id)
+ .expect("Topic not found by numeric ID")
+ }
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
- *self.index.borrow().get(&key).expect("Topic not found")
+ *self
+ .index
+ .borrow()
+ .get(&key)
+ .expect("Topic not found by name")
}
}
}
diff --git a/core/server/src/streaming/partitions/partition.rs b/core/server/src/streaming/partitions/partition.rs
index 7a07b8f33..bc06416d4 100644
--- a/core/server/src/streaming/partitions/partition.rs
+++ b/core/server/src/streaming/partitions/partition.rs
@@ -157,7 +157,7 @@ impl Partition {
}
}
- pub fn stats(&self) -> &PartitionStats {
+ pub fn stats(&self) -> &Arc<PartitionStats> {
&self.stats
}
}
diff --git a/core/server/src/streaming/topics/topic.rs b/core/server/src/streaming/topics/topic.rs
index 348dc3c8a..5a3aa1ee9 100644
--- a/core/server/src/streaming/topics/topic.rs
+++ b/core/server/src/streaming/topics/topic.rs
@@ -121,7 +121,7 @@ impl Topic {
&mut self.root
}
- pub fn stats(&self) -> &TopicStats {
+ pub fn stats(&self) -> &Arc<TopicStats> {
&self.stats
}
}
diff --git a/core/server/src/tcp/tcp_listener.rs b/core/server/src/tcp/tcp_listener.rs
index c68547301..82b539f9d 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -20,7 +20,6 @@ use crate::configs::tcp::TcpSocketConfig;
use crate::shard::IggyShard;
use crate::shard::task_registry::{ShutdownToken, TaskRegistry};
-use crate::shard::transmission::event::ShardEvent;
use crate::tcp::connection_handler::{ConnectionAction, handle_connection, handle_error};
use compio::net::{TcpListener, TcpOpts};
use err_trail::ErrContext;
@@ -72,7 +71,7 @@ pub async fn start(
if shard.id != 0 && addr.port() == 0 {
info!("Waiting for TCP address from shard 0...");
loop {
- if let Some(bound_addr) = shard.tcp_bound_address.get() {
+ if let Some(bound_addr) = shard.shared_metadata.load().bound_addresses.tcp {
addr = bound_addr;
info!("Received TCP address: {}", addr);
break;
@@ -94,19 +93,14 @@ pub async fn start(
info!("{} server has started on: {:?}", server_name, actual_addr);
if shard.id == 0 {
- // Store bound address locally
+ // Store in Cell for config_writer backward compat
shard.tcp_bound_address.set(Some(actual_addr));
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_tcp_address(actual_addr);
if addr.port() == 0 {
// Notify config writer on shard 0
let _ = shard.config_writer_notify.try_send(());
-
- // Broadcast to other shards for SO_REUSEPORT binding
- let event = ShardEvent::AddressBound {
- protocol: TransportProtocol::Tcp,
- address: actual_addr,
- };
- shard.broadcast_event_to_all_shards(event).await?;
}
}
diff --git a/core/server/src/tcp/tcp_tls_listener.rs b/core/server/src/tcp/tcp_tls_listener.rs
index 50a8c1d19..ef257b05b 100644
--- a/core/server/src/tcp/tcp_tls_listener.rs
+++ b/core/server/src/tcp/tcp_tls_listener.rs
@@ -19,7 +19,6 @@
use crate::configs::tcp::TcpSocketConfig;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
-use crate::shard::transmission::event::ShardEvent;
use crate::tcp::connection_handler::{handle_connection, handle_error};
use compio::net::{TcpListener, TcpOpts};
use compio_tls::TlsAcceptor;
@@ -46,7 +45,7 @@ pub(crate) async fn start(
if shard.id != 0 && addr.port() == 0 {
info!("Waiting for TCP address from shard 0...");
loop {
- if let Some(bound_addr) = shard.tcp_bound_address.get() {
+ if let Some(bound_addr) = shard.shared_metadata.load().bound_addresses.tcp {
addr = bound_addr;
info!("Received TCP address: {}", addr);
break;
@@ -68,16 +67,14 @@ pub(crate) async fn start(
})?;
if shard.id == 0 {
+ // Store in Cell for config_writer backward compat
shard.tcp_bound_address.set(Some(actual_addr));
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_tcp_address(actual_addr);
+
if addr.port() == 0 {
// Notify config writer on shard 0
let _ = shard.config_writer_notify.try_send(());
-
- let event = ShardEvent::AddressBound {
- protocol: TransportProtocol::Tcp,
- address: actual_addr,
- };
- shard.broadcast_event_to_all_shards(event).await?;
}
}
diff --git a/core/server/src/websocket/websocket_listener.rs b/core/server/src/websocket/websocket_listener.rs
index 7b691c61b..d17273cb6 100644
--- a/core/server/src/websocket/websocket_listener.rs
+++ b/core/server/src/websocket/websocket_listener.rs
@@ -19,7 +19,6 @@
use crate::configs::websocket::WebSocketConfig;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
-use crate::shard::transmission::event::ShardEvent;
use crate::websocket::connection_handler::{handle_connection, handle_error};
use compio::net::TcpListener;
use compio_net::TcpOpts;
@@ -58,7 +57,7 @@ pub async fn start(
if shard.id != 0 && addr.port() == 0 {
info!("Waiting for WebSocket address from shard 0...");
loop {
- if let Some(bound_addr) = shard.websocket_bound_address.get() {
+ if let Some(bound_addr) = shard.shared_metadata.load().bound_addresses.websocket {
addr = bound_addr;
info!("Received WebSocket address from shard 0: {}", addr);
break;
@@ -78,25 +77,16 @@ pub async fn start(
let local_addr = listener.local_addr().unwrap();
info!("{} has started on: ws://{}", "WebSocket Server", local_addr);
- // Notify shard about the bound address
- let event = ShardEvent::AddressBound {
- protocol: TransportProtocol::WebSocket,
- address: local_addr,
- };
-
if shard.id == 0 {
- // Store bound address locally first
+ // Store in Cell for config_writer backward compat
shard.websocket_bound_address.set(Some(local_addr));
-
- if addr.port() == 0 {
- // Broadcast to other shards for SO_REUSEPORT binding
- shard.broadcast_event_to_all_shards(event).await?;
- }
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_websocket_address(local_addr);
+ // Notify config_writer that WebSocket is bound
+ let _ = shard.config_writer_notify.try_send(());
} else {
- // Non-shard0 just handles the event locally
- crate::shard::handlers::handle_event(&shard, event)
- .await
- .ok();
+ // Non-shard-0 just updates local Cell
+ shard.websocket_bound_address.set(Some(local_addr));
}
let ws_config = config.to_tungstenite_config();
diff --git a/core/server/src/websocket/websocket_tls_listener.rs b/core/server/src/websocket/websocket_tls_listener.rs
index 3743162fb..45c080da1 100644
--- a/core/server/src/websocket/websocket_tls_listener.rs
+++ b/core/server/src/websocket/websocket_tls_listener.rs
@@ -19,7 +19,6 @@
use crate::configs::websocket::WebSocketConfig;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
-use crate::shard::transmission::event::ShardEvent;
use crate::websocket::connection_handler::{handle_connection, handle_error};
use compio::net::TcpListener;
use compio_net::TcpOpts;
@@ -63,7 +62,7 @@ pub async fn start(
if shard.id != 0 && addr.port() == 0 {
info!("Waiting for WebSocket TLS address from shard 0...");
loop {
- if let Some(bound_addr) = shard.websocket_bound_address.get() {
+ if let Some(bound_addr) = shard.shared_metadata.load().bound_addresses.websocket {
addr = bound_addr;
info!("Received WebSocket TLS address from shard 0: {}", addr);
break;
@@ -81,25 +80,16 @@ pub async fn start(
let local_addr = listener.local_addr().unwrap();
- // Notify shard about the bound address
- let event = ShardEvent::AddressBound {
- protocol: TransportProtocol::WebSocket,
- address: local_addr,
- };
-
if shard.id == 0 {
- // Store bound address locally first
+ // Store in Cell for config_writer backward compat
shard.websocket_bound_address.set(Some(local_addr));
-
- if addr.port() == 0 {
- // Broadcast to other shards for SO_REUSEPORT binding
- shard.broadcast_event_to_all_shards(event).await?;
- }
+ // Store in SharedMetadata (all shards see this immediately via ArcSwap)
+ shard.shared_metadata.set_websocket_address(local_addr);
+ // Notify config_writer that WebSocket is bound
+ let _ = shard.config_writer_notify.try_send(());
} else {
- // Non-shard0 just handles the event locally
- crate::shard::handlers::handle_event(&shard, event)
- .await
- .ok();
+ // Non-shard-0 just updates local Cell
+ shard.websocket_bound_address.set(Some(local_addr));
}
// Ensure rustls crypto provider is installed
(iggy) 03/04: commands
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch arc-swap
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 058ae50ce4b0a4d785d5d12708e898394e603658
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Dec 29 14:26:56 2025 +0100
commands
---
.../create_consumer_group_handler.rs | 8 -
.../delete_consumer_group_handler.rs | 8 -
.../partitions/create_partitions_handler.rs | 7 -
.../partitions/delete_partitions_handler.rs | 10 +-
.../create_personal_access_token_handler.rs | 5 -
.../delete_personal_access_token_handler.rs | 7 -
.../handlers/streams/create_stream_handler.rs | 7 -
.../handlers/streams/delete_stream_handler.rs | 7 -
.../handlers/streams/purge_stream_handler.rs | 6 +-
.../handlers/streams/update_stream_handler.rs | 6 -
.../binary/handlers/topics/create_topic_handler.rs | 15 +-
.../binary/handlers/topics/delete_topic_handler.rs | 8 -
.../binary/handlers/topics/purge_topic_handler.rs | 6 -
.../binary/handlers/topics/update_topic_handler.rs | 12 --
.../handlers/users/change_password_handler.rs | 8 -
.../binary/handlers/users/create_user_handler.rs | 11 -
.../binary/handlers/users/delete_user_handler.rs | 3 -
.../handlers/users/update_permissions_handler.rs | 6 -
.../binary/handlers/users/update_user_handler.rs | 8 -
core/server/src/http/consumer_groups.rs | 37 ----
core/server/src/http/partitions.rs | 41 +---
core/server/src/http/streams.rs | 70 +------
core/server/src/http/topics.rs | 114 ++--------
core/server/src/http/users.rs | 89 --------
core/server/src/shard/handlers.rs | 231 +--------------------
core/server/src/shard/transmission/event.rs | 105 +---------
26 files changed, 33 insertions(+), 802 deletions(-)
diff --git a/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
index 0c6e9a763..fed1f0a6f 100644
--- a/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::consumer_groups::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::state::models::CreateConsumerGroupWithId;
@@ -57,13 +56,6 @@ impl ServerCommandHandler for CreateConsumerGroup {
)?;
let cg_id = cg.id();
- let event = ShardEvent::CreatedConsumerGroup {
- stream_id: self.stream_id.clone(),
- topic_id: self.topic_id.clone(),
- cg,
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
let stream_id = self.stream_id.clone();
let topic_id = self.topic_id.clone();
shard
diff --git a/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
index 1da98ac44..3c2800fe7 100644
--- a/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::consumer_groups::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::streaming::polling_consumer::ConsumerGroupId;
@@ -102,13 +101,6 @@ impl ServerCommandHandler for DeleteConsumerGroup {
)
})?;
- let event = ShardEvent::DeletedConsumerGroup {
- id: cg_id,
- stream_id: self.stream_id.clone(),
- topic_id: self.topic_id.clone(),
- group_id: self.group_id.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
let stream_id = self.stream_id.clone();
let topic_id = self.topic_id.clone();
shard
diff --git a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
index 0ead0c4ad..3bd65d7bf 100644
--- a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::partitions::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
@@ -62,12 +61,6 @@ impl ServerCommandHandler for CreatePartitions {
)
.await?;
let partition_ids = partitions.iter().map(|p| p.id()).collect::<Vec<_>>();
- let event = ShardEvent::CreatedPartitions {
- stream_id: self.stream_id.clone(),
- topic_id: self.topic_id.clone(),
- partitions,
- };
- shard.broadcast_event_to_all_shards(event).await?;
shard.streams.with_topic_by_id_mut(
&self.stream_id,
diff --git a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
index fe6740136..6e2045493 100644
--- a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::partitions::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
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 +52,7 @@ impl ServerCommandHandler for DeletePartitions {
// Acquire partition lock to serialize filesystem operations
let _partition_guard = shard.fs_locks.partition_lock.lock().await;
- let deleted_partition_ids = shard
+ let _deleted_partition_ids = shard
.delete_partitions(
session,
&self.stream_id,
@@ -61,13 +60,6 @@ impl ServerCommandHandler for DeletePartitions {
self.partitions_count,
)
.await?;
- let event = ShardEvent::DeletedPartitions {
- stream_id: self.stream_id.clone(),
- topic_id: self.topic_id.clone(),
- partitions_count: self.partitions_count,
- partition_ids: deleted_partition_ids,
- };
- shard.broadcast_event_to_all_shards(event).await?;
let remaining_partition_ids = shard.streams.with_topic_by_id(
&self.stream_id,
diff --git a/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
index c52ddd4eb..6a0ec00ed 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
@@ -24,7 +24,6 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::state::models::CreatePersonalAccessTokenWithHash;
use crate::streaming::session::Session;
@@ -60,10 +59,6 @@ impl ServerCommandHandler for CreatePersonalAccessToken {
})?;
let bytes = mapper::map_raw_pat(&token);
let hash = personal_access_token.token.to_string();
- let event = ShardEvent::CreatedPersonalAccessToken {
- personal_access_token: personal_access_token.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
shard
.state
diff --git a/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
index 9a588ccbe..19b472fb3 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
@@ -53,13 +53,6 @@ impl ServerCommandHandler for DeletePersonalAccessToken {
"{COMPONENT} (error: {error}) - failed to delete personal access token with name: {token_name}, session: {session}"
)})?;
- // Broadcast the event to other shards
- let event = crate::shard::transmission::event::ShardEvent::DeletedPersonalAccessToken {
- user_id: session.get_user_id(),
- name: self.name.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
shard
.state
.apply(
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 e56f87e64..83990d9eb 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -24,7 +24,6 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
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,
@@ -77,12 +76,6 @@ impl ServerCommandHandler for CreateStream {
let stream = shard.create_stream(session, name).await?;
let created_stream_id = stream.id();
- let event = ShardEvent::CreatedStream {
- id: created_stream_id,
- stream: stream.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
let response = shard
.streams
.with_components_by_id(created_stream_id, |(root, stats)| {
diff --git a/core/server/src/binary/handlers/streams/delete_stream_handler.rs b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
index 76bc9f8ca..c1ae46336 100644
--- a/core/server/src/binary/handlers/streams/delete_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
@@ -22,7 +22,6 @@ use crate::binary::command::{
use crate::binary::handlers::streams::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
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,
@@ -83,12 +82,6 @@ impl ServerCommandHandler for DeleteStream {
stream.id()
);
- let event = ShardEvent::DeletedStream {
- id: stream.id(),
- stream_id: stream_id.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
shard
.state
.apply(session.get_user_id(), &EntryCommand::DeleteStream(DeleteStream { stream_id: stream_id.clone() }))
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 b8d3cfc0c..fd26111e7 100644
--- a/core/server/src/binary/handlers/streams/purge_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::streams::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -54,10 +53,7 @@ impl ServerCommandHandler for PurgeStream {
.with_error(|error| {
format!("{COMPONENT} (error: {error}) - failed to purge stream with id: {stream_id}, session: {session}")
})?;
- let event = ShardEvent::PurgedStream {
- stream_id: self.stream_id.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
+
shard
.state
.apply(session.get_user_id(), &EntryCommand::PurgeStream(self))
diff --git a/core/server/src/binary/handlers/streams/update_stream_handler.rs b/core/server/src/binary/handlers/streams/update_stream_handler.rs
index c080412a7..7168cae2d 100644
--- a/core/server/src/binary/handlers/streams/update_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/update_stream_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::streams::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -54,11 +53,6 @@ impl ServerCommandHandler for UpdateStream {
format!("{COMPONENT} (error: {error}) - failed to update stream with id: {stream_id}, session: {session}")
})?;
- let event = ShardEvent::UpdatedStream {
- stream_id: self.stream_id.clone(),
- name: self.name.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
shard
.state
.apply(session.get_user_id(), &EntryCommand::UpdateStream(self))
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 b5b5c96a2..b15b9c244 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -24,7 +24,6 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
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,
@@ -109,13 +108,7 @@ impl ServerCommandHandler for CreateTopic {
.with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
let topic_id = topic.id();
- let event = ShardEvent::CreatedTopic {
- stream_id: stream_id.clone(),
- topic,
- };
-
- shard.broadcast_event_to_all_shards(event).await?;
- let partitions = shard
+ shard
.create_partitions(
session,
&stream_id,
@@ -123,12 +116,6 @@ impl ServerCommandHandler for CreateTopic {
partitions_count,
)
.await?;
- let event = ShardEvent::CreatedPartitions {
- stream_id: stream_id.clone(),
- topic_id: Identifier::numeric(topic_id as u32).unwrap(),
- partitions,
- };
- shard.broadcast_event_to_all_shards(event).await?;
let response = shard.streams.with_topic_by_id(
&stream_id,
diff --git a/core/server/src/binary/handlers/topics/delete_topic_handler.rs b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
index ea3d21cdd..59b8394b7 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::topics::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
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,
@@ -91,13 +90,6 @@ impl ServerCommandHandler for DeleteTopic {
stream_id_num
);
- let event = ShardEvent::DeletedTopic {
- id: topic_id_num,
- stream_id: stream_id.clone(),
- topic_id: topic_id.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
shard
.state
.apply(session.get_user_id(), &EntryCommand::DeleteTopic(self))
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 2cea843ed..8c725f960 100644
--- a/core/server/src/binary/handlers/topics/purge_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
@@ -58,12 +58,6 @@ impl ServerCommandHandler for PurgeTopic {
)
})?;
- let event = crate::shard::transmission::event::ShardEvent::PurgedTopic {
- stream_id: self.stream_id.clone(),
- topic_id: self.topic_id.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
shard
.state
.apply(session.get_user_id(), &EntryCommand::PurgeTopic(self))
diff --git a/core/server/src/binary/handlers/topics/update_topic_handler.rs b/core/server/src/binary/handlers/topics/update_topic_handler.rs
index 6dfa2281d..0f4ae3a8a 100644
--- a/core/server/src/binary/handlers/topics/update_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/update_topic_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::topics::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
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,
@@ -117,17 +116,6 @@ impl ServerCommandHandler for UpdateTopic {
.streams
.with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
- let event = ShardEvent::UpdatedTopic {
- stream_id: stream_id.clone(),
- topic_id: topic_id.clone(),
- name: name.clone(),
- message_expiry: self.message_expiry,
- compression_algorithm: self.compression_algorithm,
- max_topic_size: self.max_topic_size,
- replication_factor: self.replication_factor,
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
shard
.state
.apply(session.get_user_id(), &EntryCommand::UpdateTopic(self))
diff --git a/core/server/src/binary/handlers/users/change_password_handler.rs b/core/server/src/binary/handlers/users/change_password_handler.rs
index dc512e172..77e4ef47d 100644
--- a/core/server/src/binary/handlers/users/change_password_handler.rs
+++ b/core/server/src/binary/handlers/users/change_password_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::users::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use crate::streaming::utils::crypto;
@@ -67,13 +66,6 @@ impl ServerCommandHandler for ChangePassword {
info!("Changed password for user with ID: {}.", self.user_id);
- let event = ShardEvent::ChangedPassword {
- user_id: self.user_id.clone(),
- current_password: self.current_password.clone(),
- new_password: self.new_password.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
// For the security of the system, we hash the password before storing it in metadata.
shard
.state
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 0caa873f1..1b5cc190e 100644
--- a/core/server/src/binary/handlers/users/create_user_handler.rs
+++ b/core/server/src/binary/handlers/users/create_user_handler.rs
@@ -23,7 +23,6 @@ use crate::binary::handlers::users::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
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,
@@ -91,16 +90,6 @@ impl ServerCommandHandler for CreateUser {
})?;
let user_id = user.id;
-
- let event = ShardEvent::CreatedUser {
- user_id,
- username: username.clone(),
- password: password.clone(),
- status,
- permissions: permissions.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
let response = mapper::map_user(&user);
shard
diff --git a/core/server/src/binary/handlers/users/delete_user_handler.rs b/core/server/src/binary/handlers/users/delete_user_handler.rs
index 55b0f103d..42b74de87 100644
--- a/core/server/src/binary/handlers/users/delete_user_handler.rs
+++ b/core/server/src/binary/handlers/users/delete_user_handler.rs
@@ -25,7 +25,6 @@ use crate::binary::handlers::users::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
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,
@@ -81,8 +80,6 @@ impl ServerCommandHandler for DeleteUser {
})?;
info!("Deleted user: {} with ID: {}.", user.username, user.id);
- let event = ShardEvent::DeletedUser { user_id };
- shard.broadcast_event_to_all_shards(event).await?;
shard
.state
diff --git a/core/server/src/binary/handlers/users/update_permissions_handler.rs b/core/server/src/binary/handlers/users/update_permissions_handler.rs
index 948ce8e41..4100da70b 100644
--- a/core/server/src/binary/handlers/users/update_permissions_handler.rs
+++ b/core/server/src/binary/handlers/users/update_permissions_handler.rs
@@ -25,7 +25,6 @@ use crate::binary::handlers::users::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -56,11 +55,6 @@ impl ServerCommandHandler for UpdatePermissions {
self.user_id
))?;
info!("Updated permissions for user with ID: {}.", self.user_id);
- let event = ShardEvent::UpdatedPermissions {
- user_id: self.user_id.clone(),
- permissions: self.permissions.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
shard
.state
diff --git a/core/server/src/binary/handlers/users/update_user_handler.rs b/core/server/src/binary/handlers/users/update_user_handler.rs
index f4e12a0f5..fae459941 100644
--- a/core/server/src/binary/handlers/users/update_user_handler.rs
+++ b/core/server/src/binary/handlers/users/update_user_handler.rs
@@ -25,7 +25,6 @@ use crate::binary::handlers::users::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::shard::IggyShard;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -66,13 +65,6 @@ impl ServerCommandHandler for UpdateUser {
info!("Updated user: {} with ID: {}.", user.username, user.id);
- let event = ShardEvent::UpdatedUser {
- user_id: self.user_id.clone(),
- username: self.username.clone(),
- status: self.status,
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
let user_id = self.user_id.clone();
shard
.state
diff --git a/core/server/src/http/consumer_groups.rs b/core/server/src/http/consumer_groups.rs
index e1e9b2f7e..278425796 100644
--- a/core/server/src/http/consumer_groups.rs
+++ b/core/server/src/http/consumer_groups.rs
@@ -180,24 +180,6 @@ async fn create_consumer_group(
let group_id = consumer_group.id();
- // Send event for consumer group creation
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::CreatedConsumerGroup {
- stream_id: command.stream_id.clone(),
- topic_id: command.topic_id.clone(),
- cg: consumer_group.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
// Get the created consumer group details
let group_id_identifier = Identifier::numeric(group_id as u32).unwrap();
let consumer_group_details = state.shard.shard().streams.with_consumer_group_by_id(
@@ -296,25 +278,6 @@ async fn delete_consumer_group(
)
})?;
- // Send event for consumer group deletion
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::DeletedConsumerGroup {
- id: cg_id,
- stream_id: identifier_stream_id.clone(),
- topic_id: identifier_topic_id.clone(),
- group_id: identifier_group_id.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
// Apply state change
let entry_command = EntryCommand::DeleteConsumerGroup(DeleteConsumerGroup {
stream_id: identifier_stream_id,
diff --git a/core/server/src/http/partitions.rs b/core/server/src/http/partitions.rs
index 46c842761..4e5f4e97b 100644
--- a/core/server/src/http/partitions.rs
+++ b/core/server/src/http/partitions.rs
@@ -20,7 +20,6 @@ use crate::http::COMPONENT;
use crate::http::error::CustomError;
use crate::http::jwt::json_web_token::Identity;
use crate::http::shared::AppState;
-use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use axum::extract::{Path, Query, State};
@@ -59,7 +58,7 @@ async fn create_partitions(
let _parititon_guard = state.shard.shard().fs_locks.partition_lock.lock().await;
let session = Session::stateless(identity.user_id, identity.ip_address);
- let partitions = SendWrapper::new(state.shard.shard().create_partitions(
+ let _partitions = SendWrapper::new(state.shard.shard().create_partitions(
&session,
&command.stream_id,
&command.topic_id,
@@ -67,24 +66,6 @@ async fn create_partitions(
))
.await?;
- let broadcast_future = SendWrapper::new(async {
- let shard = state.shard.shard();
-
- let event = ShardEvent::CreatedPartitions {
- stream_id: command.stream_id.clone(),
- topic_id: command.topic_id.clone(),
- partitions,
- };
- let _responses = shard.broadcast_event_to_all_shards(event).await;
- Ok::<(), CustomError>(())
- });
-
- broadcast_future.await
- .with_error(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to broadcast partition events, stream ID: {stream_id}, topic ID: {topic_id}"
- )
- })?;
let command = EntryCommand::CreatePartitions(command);
let state_future =
SendWrapper::new(state.shard.shard().state.apply(identity.user_id, &command));
@@ -112,7 +93,7 @@ async fn delete_partitions(
query.validate()?;
let session = Session::stateless(identity.user_id, identity.ip_address);
- let deleted_partition_ids = {
+ {
let delete_future = SendWrapper::new(state.shard.shard().delete_partitions(
&session,
&query.stream_id,
@@ -127,24 +108,6 @@ async fn delete_partitions(
})?
};
- // Send event for partition deletion
- {
- let broadcast_future = SendWrapper::new(async {
- let event = ShardEvent::DeletedPartitions {
- stream_id: query.stream_id.clone(),
- topic_id: query.topic_id.clone(),
- partitions_count: query.partitions_count,
- partition_ids: deleted_partition_ids,
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
let command = EntryCommand::DeletePartitions(DeletePartitions {
stream_id: query.stream_id.clone(),
topic_id: query.topic_id.clone(),
diff --git a/core/server/src/http/streams.rs b/core/server/src/http/streams.rs
index 2a66ea1d1..0c49feab3 100644
--- a/core/server/src/http/streams.rs
+++ b/core/server/src/http/streams.rs
@@ -117,20 +117,6 @@ async fn create_stream(
let created_stream_id = stream.root().id();
- // Send event for stream creation - inlined from wrapper
- {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::CreatedStream {
- id: created_stream_id,
- stream,
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- }
-
// Apply state change using wrapper method
let entry_command = EntryCommand::CreateStream(CreateStreamWithId {
stream_id: created_stream_id as u32,
@@ -189,23 +175,6 @@ async fn update_stream(
)
})?;
- // Send event for stream update
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::UpdatedStream {
- stream_id: command.stream_id.clone(),
- name: command.name.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
// Apply state change using wrapper method
let entry_command = EntryCommand::UpdateStream(command);
state.shard.apply_state(identity.user_id, &entry_command).await.with_error(|error| {
@@ -232,8 +201,8 @@ async fn delete_stream(
let session = Session::stateless(identity.user_id, identity.ip_address);
let _stream_guard = state.shard.shard().fs_locks.stream_lock.lock().await;
- // Delete stream and get the stream entity
- let stream = {
+ // Delete stream
+ {
let future = SendWrapper::new(
state
.shard
@@ -246,25 +215,6 @@ async fn delete_stream(
format!("{COMPONENT} (error: {error}) - failed to delete stream with ID: {stream_id}",)
})?;
- let stream_id_numeric = stream.root().id();
-
- // Send event for stream deletion
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::DeletedStream {
- id: stream_id_numeric,
- stream_id: identifier_stream_id.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
// Apply state change using wrapper method
let entry_command = EntryCommand::DeleteStream(DeleteStream {
stream_id: identifier_stream_id,
@@ -304,22 +254,6 @@ async fn purge_stream(
)
})?;
- // Send event for stream purge
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::PurgedStream {
- stream_id: identifier_stream_id.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
// Apply state change using wrapper method
let entry_command = EntryCommand::PurgeStream(PurgeStream {
stream_id: identifier_stream_id,
diff --git a/core/server/src/http/topics.rs b/core/server/src/http/topics.rs
index 551eed602..f4586a869 100644
--- a/core/server/src/http/topics.rs
+++ b/core/server/src/http/topics.rs
@@ -202,44 +202,23 @@ async fn create_topic(
let topic_id = topic.id();
- // Send events for topic creation
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
-
+ // Create partitions
+ {
let shard = state.shard.shard();
-
- let event = ShardEvent::CreatedTopic {
- stream_id: command.stream_id.clone(),
- topic,
- };
- let _responses = shard.broadcast_event_to_all_shards(event).await;
-
- // Create partitions
- let partitions = shard
- .create_partitions(
- &session,
- &command.stream_id,
- &Identifier::numeric(topic_id as u32).unwrap(),
- command.partitions_count,
- )
- .await?;
-
- let event = ShardEvent::CreatedPartitions {
- stream_id: command.stream_id.clone(),
- topic_id: Identifier::numeric(topic_id as u32).unwrap(),
- partitions,
- };
- let _responses = shard.broadcast_event_to_all_shards(event).await;
-
- Ok::<(), CustomError>(())
- });
-
- broadcast_future.await
- .with_error(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to broadcast topic events, stream ID: {stream_id}"
- )
- })?;
+ let topic_identifier = Identifier::numeric(topic_id as u32).unwrap();
+ let future = SendWrapper::new(shard.create_partitions(
+ &session,
+ &command.stream_id,
+ &topic_identifier,
+ command.partitions_count,
+ ));
+ future.await
+ }
+ .with_error(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to create partitions, stream ID: {stream_id}"
+ )
+ })?;
// Create response using the same approach as binary handler
let response = {
@@ -320,28 +299,6 @@ async fn update_topic(
|(root, _, _)| (root.message_expiry(), root.max_topic_size()),
);
- // Send event for topic update
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::UpdatedTopic {
- stream_id: command.stream_id.clone(),
- topic_id: command.topic_id.clone(),
- name: command.name.clone(),
- message_expiry: command.message_expiry,
- compression_algorithm: command.compression_algorithm,
- max_topic_size: command.max_topic_size,
- replication_factor: command.replication_factor,
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
command.message_expiry = message_expiry;
command.max_topic_size = max_topic_size;
@@ -372,7 +329,7 @@ async fn delete_topic(
let session = Session::stateless(identity.user_id, identity.ip_address);
let _topic_guard = state.shard.shard().fs_locks.topic_lock.lock().await;
- let topic = {
+ {
let future = SendWrapper::new(state.shard.shard().delete_topic(
&session,
&identifier_stream_id,
@@ -385,26 +342,6 @@ async fn delete_topic(
)
})?;
- let topic_id_numeric = topic.root().id();
-
- // Send event for topic deletion
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::DeletedTopic {
- id: topic_id_numeric,
- stream_id: identifier_stream_id.clone(),
- topic_id: identifier_topic_id.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let entry_command = EntryCommand::DeleteTopic(DeleteTopic {
stream_id: identifier_stream_id,
@@ -450,23 +387,6 @@ async fn purge_topic(
)
})?;
- // Send event for topic purge
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::PurgedTopic {
- stream_id: identifier_stream_id.clone(),
- topic_id: identifier_topic_id.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let entry_command = EntryCommand::PurgeTopic(PurgeTopic {
stream_id: identifier_stream_id,
diff --git a/core/server/src/http/users.rs b/core/server/src/http/users.rs
index 4389c6d87..ff49dae6c 100644
--- a/core/server/src/http/users.rs
+++ b/core/server/src/http/users.rs
@@ -137,26 +137,6 @@ async fn create_user(
let user_id = user.id;
let response = Json(mapper::map_user(&user));
- // Send event for user creation
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::CreatedUser {
- user_id,
- username: command.username.to_owned(),
- password: command.password.to_owned(),
- status: command.status,
- permissions: command.permissions.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let username = command.username.clone();
let entry_command = EntryCommand::CreateUser(CreateUserWithId {
@@ -212,24 +192,6 @@ async fn update_user(
format!("{COMPONENT} (error: {error}) - failed to update user, user ID: {user_id}")
})?;
- // Send event for user update
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::UpdatedUser {
- user_id: command.user_id.clone(),
- username: command.username.clone(),
- status: command.status,
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let username = command.username.clone();
let entry_command = EntryCommand::UpdateUser(command);
@@ -273,23 +235,6 @@ async fn update_permissions(
)
})?;
- // Send event for permissions update
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::UpdatedPermissions {
- user_id: command.user_id.clone(),
- permissions: command.permissions.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let entry_command = EntryCommand::UpdatePermissions(command);
let future = SendWrapper::new(
@@ -334,24 +279,6 @@ async fn change_password(
format!("{COMPONENT} (error: {error}) - failed to change password, user ID: {user_id}")
})?;
- // Send event for password change
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::ChangedPassword {
- user_id: command.user_id.clone(),
- current_password: command.current_password.clone(),
- new_password: command.new_password.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let entry_command = EntryCommand::ChangePassword(command);
let future = SendWrapper::new(
@@ -390,22 +317,6 @@ async fn delete_user(
format!("{COMPONENT} (error: {error}) - failed to delete user with ID: {user_id}")
})?;
- // Send event for user deletion
- {
- let broadcast_future = SendWrapper::new(async {
- use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::DeletedUser {
- user_id: identifier_user_id.clone(),
- };
- let _responses = state
- .shard
- .shard()
- .broadcast_event_to_all_shards(event)
- .await;
- });
- broadcast_future.await;
- }
-
{
let entry_command = EntryCommand::DeleteUser(DeleteUser {
user_id: identifier_user_id,
diff --git a/core/server/src/shard/handlers.rs b/core/server/src/shard/handlers.rs
index 81c54a98a..c1aed5c92 100644
--- a/core/server/src/shard/handlers.rs
+++ b/core/server/src/shard/handlers.rs
@@ -120,14 +120,6 @@ async fn handle_request(
let _stream_guard = shard.fs_locks.stream_lock.lock().await;
let stream = shard.create_stream(&session, name.clone()).await?;
- let created_stream_id = stream.id();
-
- let event = ShardEvent::CreatedStream {
- id: created_stream_id,
- stream: stream.clone(),
- };
-
- shard.broadcast_event_to_all_shards(event).await?;
Ok(ShardResponse::CreateStreamResponse(stream))
}
@@ -164,12 +156,8 @@ async fn handle_request(
.await?;
let topic_id = topic.id();
- let event = ShardEvent::CreatedTopic {
- stream_id: stream_id.clone(),
- topic: topic.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
- let partitions = shard
+
+ shard
.create_partitions(
&session,
&stream_id,
@@ -178,13 +166,6 @@ async fn handle_request(
)
.await?;
- let event = ShardEvent::CreatedPartitions {
- stream_id: stream_id.clone(),
- topic_id: Identifier::numeric(topic_id as u32).unwrap(),
- partitions,
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
Ok(ShardResponse::CreateTopicResponse(topic))
}
ShardRequestPayload::UpdateTopic {
@@ -215,17 +196,6 @@ async fn handle_request(
replication_factor,
)?;
- let event = ShardEvent::UpdatedTopic {
- stream_id: stream_id.clone(),
- topic_id: topic_id.clone(),
- name,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- };
- shard.broadcast_event_to_all_shards(event).await?;
-
Ok(ShardResponse::UpdateTopicResponse)
}
ShardRequestPayload::DeleteTopic {
@@ -242,14 +212,6 @@ async fn handle_request(
let _topic_guard = shard.fs_locks.topic_lock.lock().await;
let topic = shard.delete_topic(&session, &stream_id, &topic_id).await?;
- let topic_id_num = topic.root().id();
-
- let event = ShardEvent::DeletedTopic {
- id: topic_id_num,
- stream_id: stream_id.clone(),
- topic_id: topic_id.clone(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
Ok(ShardResponse::DeleteTopicResponse(topic))
}
@@ -270,16 +232,6 @@ async fn handle_request(
let user =
shard.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(),
- };
- shard.broadcast_event_to_all_shards(event).await?;
Ok(ShardResponse::CreateUserResponse(user))
}
ShardRequestPayload::GetStats { .. } => {
@@ -299,8 +251,6 @@ async fn handle_request(
);
let _user_guard = shard.fs_locks.user_lock.lock().await;
let user = shard.delete_user(&session, &user_id)?;
- let event = ShardEvent::DeletedUser { user_id };
- shard.broadcast_event_to_all_shards(event).await?;
Ok(ShardResponse::DeletedUser(user))
}
ShardRequestPayload::DeleteStream { user_id, stream_id } => {
@@ -312,11 +262,6 @@ async fn handle_request(
);
let _stream_guard = shard.fs_locks.stream_lock.lock().await;
let stream = shard.delete_stream(&session, &stream_id).await?;
- let event = ShardEvent::DeletedStream {
- id: stream.id(),
- stream_id,
- };
- shard.broadcast_event_to_all_shards(event).await?;
Ok(ShardResponse::DeleteStreamResponse(stream))
}
ShardRequestPayload::SocketTransfer {
@@ -402,86 +347,15 @@ async fn handle_request(
pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<(), IggyError> {
match event {
- ShardEvent::DeletedPartitions {
- stream_id,
- topic_id,
- partitions_count,
- partition_ids,
- } => {
- shard.delete_partitions_bypass_auth(
- &stream_id,
- &topic_id,
- partitions_count,
- partition_ids,
- )?;
- Ok(())
- }
- ShardEvent::UpdatedStream { stream_id, name } => {
- shard.update_stream_bypass_auth(&stream_id, &name)?;
- Ok(())
- }
- ShardEvent::PurgedStream { stream_id } => {
- shard.purge_stream_bypass_auth(&stream_id).await?;
- Ok(())
- }
- ShardEvent::PurgedTopic {
+ ShardEvent::FlushUnsavedBuffer {
stream_id,
topic_id,
+ partition_id,
+ fsync,
} => {
- shard.purge_topic_bypass_auth(&stream_id, &topic_id).await?;
- Ok(())
- }
- ShardEvent::CreatedUser {
- user_id,
- username,
- password,
- status,
- permissions,
- } => {
- shard.create_user_bypass_auth(
- user_id,
- &username,
- &password,
- status,
- permissions.clone(),
- )?;
- Ok(())
- }
- ShardEvent::DeletedUser { user_id } => {
- shard.delete_user_bypass_auth(&user_id)?;
- Ok(())
- }
- ShardEvent::ChangedPassword {
- user_id,
- current_password,
- new_password,
- } => {
- shard.change_password_bypass_auth(&user_id, ¤t_password, &new_password)?;
- Ok(())
- }
- ShardEvent::CreatedPersonalAccessToken {
- personal_access_token,
- } => {
- shard.create_personal_access_token_bypass_auth(personal_access_token.to_owned())?;
- Ok(())
- }
- ShardEvent::DeletedPersonalAccessToken { user_id, name } => {
- shard.delete_personal_access_token_bypass_auth(user_id, &name)?;
- Ok(())
- }
- ShardEvent::UpdatedUser {
- user_id,
- username,
- status,
- } => {
- shard.update_user_bypass_auth(&user_id, username.to_owned(), status)?;
- Ok(())
- }
- ShardEvent::UpdatedPermissions {
- user_id,
- permissions,
- } => {
- shard.update_permissions_bypass_auth(&user_id, permissions.to_owned())?;
+ shard
+ .flush_unsaved_buffer_base(&stream_id, &topic_id, partition_id, fsync)
+ .await?;
Ok(())
}
ShardEvent::AddressBound { protocol, address } => {
@@ -509,94 +383,5 @@ pub async fn handle_event(shard: &Rc<IggyShard>, event: ShardEvent) -> Result<()
}
Ok(())
}
- ShardEvent::CreatedStream { id: _, stream } => {
- // Note: Slab::insert() returns the next available slot, not the stream's embedded ID.
- // The stream already has the correct ID from shard 0, which is preserved after insert.
- shard.create_stream_bypass_auth(stream);
- Ok(())
- }
- ShardEvent::DeletedStream { id, stream_id } => {
- let stream = shard.delete_stream_bypass_auth(&stream_id);
- assert_eq!(stream.id(), id);
-
- Ok(())
- }
- ShardEvent::CreatedTopic { stream_id, topic } => {
- // Note: Slab::insert() returns the next available slot, not the topic's embedded ID.
- // The topic already has the correct ID from shard 0, which is preserved after insert.
- shard.create_topic_bypass_auth(&stream_id, topic.clone());
- Ok(())
- }
- ShardEvent::CreatedPartitions {
- stream_id,
- topic_id,
- partitions,
- } => {
- shard
- .create_partitions_bypass_auth(&stream_id, &topic_id, partitions)
- .await?;
- Ok(())
- }
- ShardEvent::DeletedTopic {
- id,
- stream_id,
- topic_id,
- } => {
- let topic = shard.delete_topic_bypass_auth(&stream_id, &topic_id);
- assert_eq!(topic.id(), id);
- Ok(())
- }
- ShardEvent::UpdatedTopic {
- stream_id,
- topic_id,
- name,
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- } => {
- shard.update_topic_bypass_auth(
- &stream_id,
- &topic_id,
- name.clone(),
- message_expiry,
- compression_algorithm,
- max_topic_size,
- replication_factor,
- )?;
- Ok(())
- }
- ShardEvent::CreatedConsumerGroup {
- stream_id,
- topic_id,
- cg,
- } => {
- // Note: Slab::insert() returns the next available slot, not the consumer group's embedded ID.
- // The consumer group already has the correct ID from shard 0.
- shard.create_consumer_group_bypass_auth(&stream_id, &topic_id, cg);
- Ok(())
- }
- ShardEvent::DeletedConsumerGroup {
- id,
- stream_id,
- topic_id,
- group_id,
- } => {
- let cg = shard.delete_consumer_group_bypass_auth(&stream_id, &topic_id, &group_id);
- assert_eq!(cg.id(), id);
-
- Ok(())
- }
- ShardEvent::FlushUnsavedBuffer {
- stream_id,
- topic_id,
- partition_id,
- fsync,
- } => {
- shard
- .flush_unsaved_buffer_base(&stream_id, &topic_id, partition_id, fsync)
- .await?;
- Ok(())
- }
}
}
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index eef75d958..3364a55aa 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -15,23 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use crate::streaming::{
- partitions::partition,
- personal_access_tokens::personal_access_token::PersonalAccessToken,
- streams::stream,
- topics::{
- consumer_group::{self},
- topic,
- },
-};
-use iggy_common::{
- CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, Permissions, TransportProtocol,
- UserStatus,
-};
+use iggy_common::{Identifier, TransportProtocol};
use std::net::SocketAddr;
use strum::Display;
-#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, Display)]
#[strum(serialize_all = "PascalCase")]
pub enum ShardEvent {
@@ -41,96 +28,6 @@ pub enum ShardEvent {
partition_id: usize,
fsync: bool,
},
- CreatedStream {
- id: usize,
- stream: stream::Stream,
- },
- DeletedStream {
- id: usize,
- stream_id: Identifier,
- },
- UpdatedStream {
- stream_id: Identifier,
- name: String,
- },
- PurgedStream {
- stream_id: Identifier,
- },
- CreatedPartitions {
- stream_id: Identifier,
- topic_id: Identifier,
- partitions: Vec<partition::Partition>,
- },
- DeletedPartitions {
- stream_id: Identifier,
- topic_id: Identifier,
- partitions_count: u32,
- partition_ids: Vec<usize>,
- },
- CreatedTopic {
- stream_id: Identifier,
- topic: topic::Topic,
- },
- CreatedConsumerGroup {
- stream_id: Identifier,
- topic_id: Identifier,
- cg: consumer_group::ConsumerGroup,
- },
- DeletedConsumerGroup {
- id: usize,
- stream_id: Identifier,
- topic_id: Identifier,
- group_id: Identifier,
- },
- UpdatedTopic {
- stream_id: Identifier,
- topic_id: Identifier,
- name: String,
- message_expiry: IggyExpiry,
- compression_algorithm: CompressionAlgorithm,
- max_topic_size: MaxTopicSize,
- replication_factor: Option<u8>,
- },
- PurgedTopic {
- stream_id: Identifier,
- topic_id: Identifier,
- },
- DeletedTopic {
- id: usize,
- stream_id: Identifier,
- topic_id: Identifier,
- },
- CreatedUser {
- user_id: u32,
- username: String,
- password: String,
- status: UserStatus,
- permissions: Option<Permissions>,
- },
- UpdatedPermissions {
- user_id: Identifier,
- permissions: Option<Permissions>,
- },
- DeletedUser {
- user_id: Identifier,
- },
- UpdatedUser {
- user_id: Identifier,
- username: Option<String>,
- status: Option<UserStatus>,
- },
- ChangedPassword {
- user_id: Identifier,
- current_password: String,
- new_password: String,
- },
- CreatedPersonalAccessToken {
- personal_access_token: PersonalAccessToken,
- },
- DeletedPersonalAccessToken {
- user_id: u32,
- name: String,
- },
AddressBound {
protocol: TransportProtocol,
address: SocketAddr,
(iggy) 02/04: cg, users, partitions
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch arc-swap
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 0d9fab57e0c824633cd1b9b435a3faa938d23b59
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Dec 29 14:12:24 2025 +0100
cg, users, partitions
---
core/server/src/shard/system/consumer_groups.rs | 17 +++++++++-
core/server/src/shard/system/partitions.rs | 10 ++++++
core/server/src/shard/system/users.rs | 43 ++++++++++++++++++++++---
3 files changed, 64 insertions(+), 6 deletions(-)
diff --git a/core/server/src/shard/system/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
index 11a9296a8..39168ece9 100644
--- a/core/server/src/shard/system/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -69,7 +69,16 @@ impl IggyShard {
topic_id,
).with_error(|error| format!("{COMPONENT} (error: {error}) - permission denied to create consumer group for user {} on stream ID: {}, topic ID: {}", session.get_user_id(), stream_id, topic_id))?;
}
- let cg = self.create_and_insert_consumer_group_mem(stream_id, topic_id, name);
+ let cg = self.create_and_insert_consumer_group_mem(stream_id, topic_id, name.clone());
+
+ // Dual-write: also update SharedMetadata
+ let partition_ids = self.streams.with_topics(stream_id, |topics| {
+ topics.with_partitions(topic_id, partitions::helpers::get_partition_ids())
+ });
+ let _ =
+ self.shared_metadata
+ .create_consumer_group(stream_id, topic_id, name, partition_ids);
+
Ok(cg)
}
@@ -131,6 +140,12 @@ impl IggyShard {
).with_error(|error| format!("{COMPONENT} (error: {error}) - permission denied to delete consumer group for user {} on stream ID: {}, topic ID: {}", session.get_user_id(), stream_id, topic_id))?;
}
let cg = self.delete_consumer_group_base(stream_id, topic_id, group_id);
+
+ // Dual-write: also update SharedMetadata
+ let _ = self
+ .shared_metadata
+ .delete_consumer_group(stream_id, topic_id, group_id);
+
Ok(cg)
}
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index b95847303..10177bab3 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -100,6 +100,11 @@ impl IggyShard {
self.metrics.increment_partitions(partitions_count);
self.metrics.increment_segments(partitions_count);
+ // Dual-write: also update SharedMetadata
+ let _ = self
+ .shared_metadata
+ .create_partitions(stream_id, topic_id, partitions_count);
+
let shards_count = self.get_available_shards_count();
for (partition_id, stats) in partitions.iter().map(|p| (p.id(), p.stats())) {
let ns = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
@@ -276,6 +281,11 @@ impl IggyShard {
parent.decrement_size_bytes(total_size_bytes);
parent.decrement_segments_count(total_segments_count);
+ // Dual-write: also update SharedMetadata
+ let _ = self
+ .shared_metadata
+ .delete_partitions(stream_id, topic_id, &deleted_ids);
+
Ok(deleted_ids)
}
diff --git a/core/server/src/shard/system/users.rs b/core/server/src/shard/system/users.rs
index 634551e39..548aaf320 100644
--- a/core/server/src/shard/system/users.rs
+++ b/core/server/src/shard/system/users.rs
@@ -105,7 +105,16 @@ impl IggyShard {
return Err(IggyError::UsersLimitReached);
}
- let user_id = self.create_user_base(username, password, status, permissions)?;
+ let user_id = self.create_user_base(username, password, status, permissions.clone())?;
+
+ // Dual-write: also update SharedMetadata
+ let _ = self.shared_metadata.create_user(
+ username.to_string(),
+ crypto::hash_password(password),
+ status,
+ permissions,
+ );
+
self.get_user(&(user_id as u32).try_into()?)
.with_error(|error| {
format!("{COMPONENT} (error: {error}) - failed to get user with id: {user_id}")
@@ -159,7 +168,12 @@ impl IggyShard {
)
})?;
- self.delete_user_base(user_id)
+ let user = self.delete_user_base(user_id)?;
+
+ // Dual-write: also update SharedMetadata
+ let _ = self.shared_metadata.delete_user(user_id);
+
+ Ok(user)
}
pub fn delete_user_bypass_auth(&self, user_id: &Identifier) -> Result<User, IggyError> {
@@ -215,7 +229,12 @@ impl IggyShard {
)
})?;
- self.update_user_base(user_id, username, status)
+ let user = self.update_user_base(user_id, username.clone(), status)?;
+
+ // Dual-write: also update SharedMetadata
+ let _ = self.shared_metadata.update_user(user_id, username, status);
+
+ Ok(user)
}
pub fn update_user_bypass_auth(
@@ -286,7 +305,14 @@ impl IggyShard {
}
}
- self.update_permissions_base(user_id, permissions)
+ self.update_permissions_base(user_id, permissions.clone())?;
+
+ // Dual-write: also update SharedMetadata
+ let _ = self
+ .shared_metadata
+ .update_permissions(user_id, permissions);
+
+ Ok(())
}
pub fn update_permissions_bypass_auth(
@@ -340,7 +366,14 @@ impl IggyShard {
}
}
- self.change_password_base(user_id, current_password, new_password)
+ self.change_password_base(user_id, current_password, new_password)?;
+
+ // Dual-write: also update SharedMetadata
+ let _ = self
+ .shared_metadata
+ .change_password(user_id, crypto::hash_password(new_password));
+
+ Ok(())
}
pub fn change_password_bypass_auth(