You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/03/05 09:07:15 UTC
(iggy) branch fix_state updated (418e985b -> e8975d53)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a change to branch fix_state
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 418e985b Fix server state by saving numeric ID for the resources
add 71716c73 fix(ci): fix protected branches checks in .asf file (#1604)
add dc97d799 chore(ci): update assignees and reviewers (#1601)
add 8952bcc8 chore(deps): bump libc from 0.2.169 to 0.2.170 (#1586)
add dd916b90 chore(ci): update CI workflows rust-cache version and key (#1608)
new e8975d53 Fix server state by saving numeric ID for the resources
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (418e985b)
\
N -- N -- N refs/heads/fix_state (e8975d53)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.asf.yaml | 2 +-
.github/dependabot.yml | 8 ++++----
.github/workflows/backwards_compatibility.yml | 2 +-
.github/workflows/coverage.yml | 2 +-
.github/workflows/performance.yml | 2 +-
.github/workflows/post_publish_server.yml | 2 +-
.github/workflows/publish_server.yml | 2 +-
.github/workflows/release_cli.yml | 7 +------
.github/workflows/test_daily.yml | 2 +-
.github/workflows/test_nightly.yml | 2 +-
.github/workflows/test_pr.yml | 6 +++---
Cargo.lock | 4 ++--
integration/Cargo.toml | 2 +-
13 files changed, 19 insertions(+), 24 deletions(-)
(iggy) 01/01: Fix server state by saving numeric ID for the resources
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch fix_state
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e8975d53420edb64319087ac65132e671087105d
Author: spetz <pi...@gmail.com>
AuthorDate: Mon Mar 3 21:14:52 2025 +0100
Fix server state by saving numeric ID for the resources
---
configs/server.toml | 2 +-
integration/tests/state/file.rs | 50 ++--
integration/tests/state/system.rs | 72 +++--
integration/tests/streaming/stream.rs | 1 -
integration/tests/streaming/topic.rs | 1 -
sdk/src/error.rs | 4 +-
.../create_consumer_group_handler.rs | 11 +-
.../handlers/streams/create_stream_handler.rs | 11 +-
.../binary/handlers/topics/create_topic_handler.rs | 7 +-
.../binary/handlers/users/create_user_handler.rs | 15 +-
server/src/http/consumer_groups.rs | 7 +-
server/src/http/streams.rs | 11 +-
server/src/http/topics.rs | 10 +-
server/src/http/users.rs | 15 +-
server/src/state/command.rs | 29 +-
server/src/state/models.rs | 324 +++++++++++++++++++--
server/src/state/system.rs | 36 +--
server/src/streaming/systems/users.rs | 6 +-
18 files changed, 485 insertions(+), 127 deletions(-)
diff --git a/configs/server.toml b/configs/server.toml
index 0ea8afe9..7c934d48 100644
--- a/configs/server.toml
+++ b/configs/server.toml
@@ -490,4 +490,4 @@ expiry = "1 m"
# Recovery configuration in case of lost data
[system.recovery]
# Controls whether streams/topics/partitions should be recreated if the expected data for existing state is missing (boolean).
-recreate_missing_state = true
+recreate_missing_state = false
diff --git a/integration/tests/state/file.rs b/integration/tests/state/file.rs
index 39196f9f..f6e4479f 100644
--- a/integration/tests/state/file.rs
+++ b/integration/tests/state/file.rs
@@ -5,6 +5,7 @@ use iggy::streams::create_stream::CreateStream;
use iggy::users::create_user::CreateUser;
use server::state::command::EntryCommand;
use server::state::entry::StateEntry;
+use server::state::models::{CreateStreamWithId, CreateUserWithId};
use server::state::State;
#[tokio::test]
@@ -23,11 +24,14 @@ async fn should_apply_single_entry() {
state.init().await.unwrap();
let user_id = 1;
- let command = EntryCommand::CreateUser(CreateUser {
- username: "test".to_string(),
- password: "secret".to_string(),
- status: Default::default(),
- permissions: None,
+ let command = EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: CreateUser {
+ username: "test".to_string(),
+ password: "secret".to_string(),
+ status: Default::default(),
+ permissions: None,
+ },
});
let command_bytes = command.to_bytes();
@@ -46,11 +50,14 @@ async fn should_apply_encrypted_entry() {
state.init().await.unwrap();
let user_id = 1;
- let command = EntryCommand::CreateUser(CreateUser {
- username: "test".to_string(),
- password: "secret".to_string(),
- status: Default::default(),
- permissions: None,
+ let command = EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: CreateUser {
+ username: "test".to_string(),
+ password: "secret".to_string(),
+ status: Default::default(),
+ permissions: None,
+ },
});
let command_bytes = command.to_bytes();
@@ -74,11 +81,14 @@ async fn should_apply_multiple_entries() {
assert_eq!(state.term(), 0);
let first_user_id = 1;
- let create_user = EntryCommand::CreateUser(CreateUser {
- username: "test".to_string(),
- password: "secret".to_string(),
- status: Default::default(),
- permissions: None,
+ let create_user = EntryCommand::CreateUser(CreateUserWithId {
+ user_id: first_user_id,
+ command: CreateUser {
+ username: "test".to_string(),
+ password: "secret".to_string(),
+ status: Default::default(),
+ permissions: None,
+ },
});
let create_user_bytes = create_user.to_bytes();
@@ -88,9 +98,13 @@ async fn should_apply_multiple_entries() {
assert_eq!(state.entries_count(), 1);
let second_user_id = 2;
- let create_stream = EntryCommand::CreateStream(CreateStream {
- stream_id: Some(1),
- name: "test".to_string(),
+ let stream_id = 1;
+ let create_stream = EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id,
+ command: CreateStream {
+ stream_id: Some(stream_id),
+ name: "test".to_string(),
+ },
});
let create_stream_bytes = create_stream.to_bytes();
diff --git a/integration/tests/state/system.rs b/integration/tests/state/system.rs
index 5c304ce8..5afe79a2 100644
--- a/integration/tests/state/system.rs
+++ b/integration/tests/state/system.rs
@@ -7,7 +7,9 @@ use iggy::topics::create_topic::CreateTopic;
use iggy::users::create_user::CreateUser;
use iggy::utils::expiry::IggyExpiry;
use server::state::command::EntryCommand;
-use server::state::models::CreatePersonalAccessTokenWithHash;
+use server::state::models::{
+ CreatePersonalAccessTokenWithHash, CreateStreamWithId, CreateTopicWithId, CreateUserWithId,
+};
use server::state::system::SystemState;
use server::state::State;
@@ -31,19 +33,21 @@ async fn should_be_initialized_based_on_state_entries() {
permissions: None,
};
+ let stream1_id = 1;
let create_stream1 = CreateStream {
- stream_id: Some(1),
+ stream_id: Some(stream1_id),
name: "stream1".to_string(),
};
let create_stream1_clone = CreateStream {
- stream_id: Some(1),
+ stream_id: Some(stream1_id),
name: "stream1".to_string(),
};
+ let topic1_id = 1;
let create_topic1 = CreateTopic {
- stream_id: create_stream1.stream_id.unwrap().try_into().unwrap(),
- topic_id: Some(1),
+ stream_id: stream1_id.try_into().unwrap(),
+ topic_id: Some(topic1_id),
partitions_count: 1,
compression_algorithm: Default::default(),
message_expiry: Default::default(),
@@ -53,8 +57,8 @@ async fn should_be_initialized_based_on_state_entries() {
};
let create_topic1_clone = CreateTopic {
- stream_id: create_stream1.stream_id.unwrap().try_into().unwrap(),
- topic_id: Some(1),
+ stream_id: stream1_id.try_into().unwrap(),
+ topic_id: Some(topic1_id),
partitions_count: 1,
compression_algorithm: Default::default(),
message_expiry: Default::default(),
@@ -63,14 +67,16 @@ async fn should_be_initialized_based_on_state_entries() {
replication_factor: None,
};
+ let stream2_id = 2;
let create_stream2 = CreateStream {
- stream_id: Some(2),
+ stream_id: Some(stream2_id),
name: "stream2".to_string(),
};
+ let topic2_id = 2;
let create_topic2 = CreateTopic {
- stream_id: create_stream2.stream_id.unwrap().try_into().unwrap(),
- topic_id: Some(2),
+ stream_id: stream2_id.try_into().unwrap(),
+ topic_id: Some(topic2_id),
partitions_count: 1,
compression_algorithm: Default::default(),
message_expiry: Default::default(),
@@ -80,13 +86,13 @@ async fn should_be_initialized_based_on_state_entries() {
};
let create_partitions = CreatePartitions {
- stream_id: create_topic1.stream_id.clone(),
- topic_id: create_topic1.topic_id.unwrap().try_into().unwrap(),
+ stream_id: stream1_id.try_into().unwrap(),
+ topic_id: topic1_id.try_into().unwrap(),
partitions_count: 2,
};
let delete_stream2 = DeleteStream {
- stream_id: create_stream2.stream_id.unwrap().try_into().unwrap(),
+ stream_id: stream2_id.try_into().unwrap(),
};
let create_personal_access_token = CreatePersonalAccessTokenWithHash {
@@ -105,23 +111,53 @@ async fn should_be_initialized_based_on_state_entries() {
hash: "hash".to_string(),
};
state
- .apply(user_id, EntryCommand::CreateUser(create_user))
+ .apply(
+ user_id,
+ EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: create_user,
+ }),
+ )
.await
.unwrap();
state
- .apply(user_id, EntryCommand::CreateStream(create_stream1))
+ .apply(
+ user_id,
+ EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id: stream1_id,
+ command: create_stream1,
+ }),
+ )
.await
.unwrap();
state
- .apply(user_id, EntryCommand::CreateTopic(create_topic1))
+ .apply(
+ user_id,
+ EntryCommand::CreateTopic(CreateTopicWithId {
+ topic_id: topic1_id,
+ command: create_topic1,
+ }),
+ )
.await
.unwrap();
state
- .apply(user_id, EntryCommand::CreateStream(create_stream2))
+ .apply(
+ user_id,
+ EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id: stream2_id,
+ command: create_stream2,
+ }),
+ )
.await
.unwrap();
state
- .apply(user_id, EntryCommand::CreateTopic(create_topic2))
+ .apply(
+ user_id,
+ EntryCommand::CreateTopic(CreateTopicWithId {
+ topic_id: topic2_id,
+ command: create_topic2,
+ }),
+ )
.await
.unwrap();
state
diff --git a/integration/tests/streaming/stream.rs b/integration/tests/streaming/stream.rs
index 0e65f621..223e09b3 100644
--- a/integration/tests/streaming/stream.rs
+++ b/integration/tests/streaming/stream.rs
@@ -61,7 +61,6 @@ async fn should_load_existing_stream_from_disk() {
name: name.clone(),
created_at: IggyTimestamp::now(),
topics: AHashMap::new(),
- current_topic_id: 0,
};
loaded_stream.load(state).await.unwrap();
diff --git a/integration/tests/streaming/topic.rs b/integration/tests/streaming/topic.rs
index 266ec149..c4f72128 100644
--- a/integration/tests/streaming/topic.rs
+++ b/integration/tests/streaming/topic.rs
@@ -117,7 +117,6 @@ async fn should_load_existing_topic_from_disk() {
max_topic_size: MaxTopicSize::ServerDefault,
replication_factor: Some(1),
created_at: Default::default(),
- current_consumer_group_id: 0,
};
loaded_topic.load(topic_state).await.unwrap();
diff --git a/sdk/src/error.rs b/sdk/src/error.rs
index 9f7f24d5..fd48e2c6 100644
--- a/sdk/src/error.rs
+++ b/sdk/src/error.rs
@@ -127,9 +127,9 @@ pub enum IggyError {
#[error("Invalid number encoding")]
InvalidNumberEncoding = 82,
#[error("Invalid boolean value")]
- InvalidBooleanValue,
+ InvalidBooleanValue = 83,
#[error("Invalid number value")]
- InvalidNumberValue,
+ InvalidNumberValue = 84,
#[error("Client with ID: {0} was not found.")]
ClientNotFound(u32) = 100,
#[error("Invalid client ID")]
diff --git a/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs b/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
index a1f13959..839cab6e 100644
--- a/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
+++ b/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
@@ -1,6 +1,7 @@
use crate::binary::mapper;
use crate::binary::{handlers::consumer_groups::COMPONENT, sender::SenderKind};
use crate::state::command::EntryCommand;
+use crate::state::models::CreateConsumerGroupWithId;
use crate::streaming::session::Session;
use crate::streaming::systems::system::SharedSystem;
use anyhow::Result;
@@ -35,25 +36,27 @@ pub async fn handle(
)
})?;
let consumer_group = consumer_group.read().await;
+ let group_id = consumer_group.group_id;
let response = mapper::map_consumer_group(&consumer_group).await;
drop(consumer_group);
let system = system.downgrade();
let stream_id = command.stream_id.clone();
let topic_id = command.topic_id.clone();
- let group_id = command.group_id;
system
.state
.apply(
session.get_user_id(),
- EntryCommand::CreateConsumerGroup(command),
+ EntryCommand::CreateConsumerGroup(CreateConsumerGroupWithId {
+ group_id,
+ command
+ }),
)
.await
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to apply create consumer group for stream_id: {}, topic_id: {}, group_id: {:?}, session: {}",
- stream_id, topic_id, group_id, session
+ "{COMPONENT} (error: {error}) - failed to apply create consumer group for stream with ID: {stream_id}, topic ID: {topic_id}, group ID: {group_id}, session: {session}",
)
})?;
sender.send_ok_response(&response).await?;
diff --git a/server/src/binary/handlers/streams/create_stream_handler.rs b/server/src/binary/handlers/streams/create_stream_handler.rs
index c32143a7..ac17ab78 100644
--- a/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -1,6 +1,7 @@
use crate::binary::mapper;
use crate::binary::{handlers::streams::COMPONENT, sender::SenderKind};
use crate::state::command::EntryCommand;
+use crate::state::models::CreateStreamWithId;
use crate::streaming::session::Session;
use crate::streaming::systems::system::SharedSystem;
use anyhow::Result;
@@ -25,16 +26,20 @@ pub async fn handle(
.await
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to create stream with id: {:?}, session: {session}",
- stream_id
+ "{COMPONENT} (error: {error}) - failed to create stream with ID: {:?}, name: {} session: {session}",
+ stream_id, command.name
)
})?;
+ let stream_id = stream.stream_id;
let response = mapper::map_stream(stream);
let system = system.downgrade();
system
.state
- .apply(session.get_user_id(), EntryCommand::CreateStream(command))
+ .apply(session.get_user_id(), EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id,
+ command
+ }))
.await
.with_error_context(|error| {
format!(
diff --git a/server/src/binary/handlers/topics/create_topic_handler.rs b/server/src/binary/handlers/topics/create_topic_handler.rs
index 36794bb3..4d2fcaac 100644
--- a/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -1,6 +1,7 @@
use crate::binary::mapper;
use crate::binary::{handlers::topics::COMPONENT, sender::SenderKind};
use crate::state::command::EntryCommand;
+use crate::state::models::CreateTopicWithId;
use crate::streaming::session::Session;
use crate::streaming::systems::system::SharedSystem;
use anyhow::Result;
@@ -38,12 +39,16 @@ pub async fn handle(
))?;
command.message_expiry = topic.message_expiry;
command.max_topic_size = topic.max_topic_size;
+ let topic_id = topic.topic_id;
let response = mapper::map_topic(topic).await;
let system = system.downgrade();
system
.state
- .apply(session.get_user_id(), EntryCommand::CreateTopic(command))
+ .apply(session.get_user_id(), EntryCommand::CreateTopic(CreateTopicWithId {
+ topic_id,
+ command
+ }))
.await
.with_error_context(|error| {
format!(
diff --git a/server/src/binary/handlers/users/create_user_handler.rs b/server/src/binary/handlers/users/create_user_handler.rs
index fec3dfd5..9cad8cce 100644
--- a/server/src/binary/handlers/users/create_user_handler.rs
+++ b/server/src/binary/handlers/users/create_user_handler.rs
@@ -1,6 +1,7 @@
use crate::binary::mapper;
use crate::binary::{handlers::users::COMPONENT, sender::SenderKind};
use crate::state::command::EntryCommand;
+use crate::state::models::CreateUserWithId;
use crate::streaming::session::Session;
use crate::streaming::systems::system::SharedSystem;
use crate::streaming::utils::crypto;
@@ -35,6 +36,7 @@ pub async fn handle(
command.username
)
})?;
+ let user_id = user.id;
let response = mapper::map_user(user);
// For the security of the system, we hash the password before storing it in metadata.
@@ -43,11 +45,14 @@ pub async fn handle(
.state
.apply(
session.get_user_id(),
- EntryCommand::CreateUser(CreateUser {
- username: command.username.to_owned(),
- password: crypto::hash_password(&command.password),
- status: command.status,
- permissions: command.permissions.clone(),
+ EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: CreateUser {
+ username: command.username.to_owned(),
+ password: crypto::hash_password(&command.password),
+ status: command.status,
+ permissions: command.permissions.clone(),
+ }
}),
)
.await
diff --git a/server/src/http/consumer_groups.rs b/server/src/http/consumer_groups.rs
index ede49d34..47d66c7c 100644
--- a/server/src/http/consumer_groups.rs
+++ b/server/src/http/consumer_groups.rs
@@ -4,6 +4,7 @@ use crate::http::mapper;
use crate::http::shared::AppState;
use crate::http::COMPONENT;
use crate::state::command::EntryCommand;
+use crate::state::models::CreateConsumerGroupWithId;
use crate::streaming::session::Session;
use axum::extract::{Path, State};
use axum::http::StatusCode;
@@ -96,13 +97,17 @@ async fn create_consumer_group(
.await
.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to create consumer group, stream ID: {}, topic ID: {}, group ID: {:?}", stream_id, topic_id, command.group_id))?;
let consumer_group = consumer_group.read().await;
+ let group_id = consumer_group.group_id;
let consumer_group_details = mapper::map_consumer_group(&consumer_group).await;
drop(consumer_group);
let system = system.downgrade();
system
.state
- .apply(identity.user_id, EntryCommand::CreateConsumerGroup(command))
+ .apply(
+ identity.user_id,
+ EntryCommand::CreateConsumerGroup(CreateConsumerGroupWithId { group_id, command }),
+ )
.await?;
Ok((StatusCode::CREATED, Json(consumer_group_details)))
diff --git a/server/src/http/streams.rs b/server/src/http/streams.rs
index 2ff6292d..c1072f35 100644
--- a/server/src/http/streams.rs
+++ b/server/src/http/streams.rs
@@ -18,6 +18,7 @@ use iggy::streams::update_stream::UpdateStream;
use iggy::validatable::Validatable;
use crate::state::command::EntryCommand;
+use crate::state::models::CreateStreamWithId;
use std::sync::Arc;
use tracing::instrument;
@@ -92,18 +93,20 @@ async fn create_stream(
command.stream_id
)
})?;
+ let stream_id = stream.stream_id;
let response = Json(mapper::map_stream(stream));
let system = system.downgrade();
- let stream_id = command.stream_id;
system
.state
- .apply(identity.user_id, EntryCommand::CreateStream(command))
+ .apply(identity.user_id, EntryCommand::CreateStream(CreateStreamWithId {
+ stream_id,
+ command
+ }))
.await
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to apply create stream, stream ID: {:?}",
- stream_id
+ "{COMPONENT} (error: {error}) - failed to apply create stream, stream ID: {stream_id}",
)
})?;
Ok(response)
diff --git a/server/src/http/topics.rs b/server/src/http/topics.rs
index 9d1f3847..3631362f 100644
--- a/server/src/http/topics.rs
+++ b/server/src/http/topics.rs
@@ -4,6 +4,7 @@ use crate::http::mapper;
use crate::http::shared::AppState;
use crate::http::COMPONENT;
use crate::state::command::EntryCommand;
+use crate::state::models::CreateTopicWithId;
use crate::streaming::session::Session;
use axum::extract::{Path, State};
use axum::http::StatusCode;
@@ -114,17 +115,20 @@ async fn create_topic(
})?;
command.message_expiry = topic.message_expiry;
command.max_topic_size = topic.max_topic_size;
+ let topic_id = topic.topic_id;
let response = Json(mapper::map_topic(topic).await);
let system = system.downgrade();
system
.state
- .apply(identity.user_id, EntryCommand::CreateTopic(command))
+ .apply(identity.user_id, EntryCommand::CreateTopic(CreateTopicWithId {
+ topic_id,
+ command
+ }))
.await
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to apply create topic, stream ID: {}",
- stream_id
+ "{COMPONENT} (error: {error}) - failed to apply create topic, stream ID: {stream_id}",
)
})?;
Ok(response)
diff --git a/server/src/http/users.rs b/server/src/http/users.rs
index e3dc10b5..e82866e8 100644
--- a/server/src/http/users.rs
+++ b/server/src/http/users.rs
@@ -5,6 +5,7 @@ use crate::http::mapper::map_generated_access_token_to_identity_info;
use crate::http::shared::AppState;
use crate::http::COMPONENT;
use crate::state::command::EntryCommand;
+use crate::state::models::CreateUserWithId;
use crate::streaming::session::Session;
use crate::streaming::utils::crypto;
use axum::extract::{Path, State};
@@ -104,6 +105,7 @@ async fn create_user(
command.username
)
})?;
+ let user_id = user.id;
let response = Json(mapper::map_user(user));
// For the security of the system, we hash the password before storing it in metadata.
@@ -112,11 +114,14 @@ async fn create_user(
.state
.apply(
identity.user_id,
- EntryCommand::CreateUser(CreateUser {
- username: command.username.clone(),
- password: crypto::hash_password(&command.password),
- status: command.status,
- permissions: command.permissions,
+ EntryCommand::CreateUser(CreateUserWithId {
+ user_id,
+ command: CreateUser {
+ username: command.username.to_owned(),
+ password: crypto::hash_password(&command.password),
+ status: command.status,
+ permissions: command.permissions.clone(),
+ },
}),
)
.await
diff --git a/server/src/state/command.rs b/server/src/state/command.rs
index 00125559..f9165ef5 100644
--- a/server/src/state/command.rs
+++ b/server/src/state/command.rs
@@ -1,4 +1,7 @@
-use crate::state::models::CreatePersonalAccessTokenWithHash;
+use crate::state::models::{
+ CreateConsumerGroupWithId, CreatePersonalAccessTokenWithHash, CreateStreamWithId,
+ CreateTopicWithId, CreateUserWithId,
+};
use bytes::{Buf, BufMut, Bytes, BytesMut};
use iggy::bytes_serializable::BytesSerializable;
use iggy::command::{
@@ -8,22 +11,18 @@ use iggy::command::{
DELETE_STREAM_CODE, DELETE_TOPIC_CODE, DELETE_USER_CODE, PURGE_STREAM_CODE, PURGE_TOPIC_CODE,
UPDATE_PERMISSIONS_CODE, UPDATE_STREAM_CODE, UPDATE_TOPIC_CODE, UPDATE_USER_CODE,
};
-use iggy::consumer_groups::create_consumer_group::CreateConsumerGroup;
use iggy::consumer_groups::delete_consumer_group::DeleteConsumerGroup;
use iggy::error::IggyError;
use iggy::partitions::create_partitions::CreatePartitions;
use iggy::partitions::delete_partitions::DeletePartitions;
use iggy::personal_access_tokens::delete_personal_access_token::DeletePersonalAccessToken;
-use iggy::streams::create_stream::CreateStream;
use iggy::streams::delete_stream::DeleteStream;
use iggy::streams::purge_stream::PurgeStream;
use iggy::streams::update_stream::UpdateStream;
-use iggy::topics::create_topic::CreateTopic;
use iggy::topics::delete_topic::DeleteTopic;
use iggy::topics::purge_topic::PurgeTopic;
use iggy::topics::update_topic::UpdateTopic;
use iggy::users::change_password::ChangePassword;
-use iggy::users::create_user::CreateUser;
use iggy::users::delete_user::DeleteUser;
use iggy::users::update_permissions::UpdatePermissions;
use iggy::users::update_user::UpdateUser;
@@ -31,19 +30,19 @@ use std::fmt::{Display, Formatter};
#[derive(Debug, PartialEq)]
pub enum EntryCommand {
- CreateStream(CreateStream),
+ CreateStream(CreateStreamWithId),
UpdateStream(UpdateStream),
DeleteStream(DeleteStream),
PurgeStream(PurgeStream),
- CreateTopic(CreateTopic),
+ CreateTopic(CreateTopicWithId),
UpdateTopic(UpdateTopic),
DeleteTopic(DeleteTopic),
PurgeTopic(PurgeTopic),
CreatePartitions(CreatePartitions),
DeletePartitions(DeletePartitions),
- CreateConsumerGroup(CreateConsumerGroup),
+ CreateConsumerGroup(CreateConsumerGroupWithId),
DeleteConsumerGroup(DeleteConsumerGroup),
- CreateUser(CreateUser),
+ CreateUser(CreateUserWithId),
UpdateUser(UpdateUser),
DeleteUser(DeleteUser),
ChangePassword(ChangePassword),
@@ -95,7 +94,7 @@ impl BytesSerializable for EntryCommand {
let length = bytes.slice(4..8).get_u32_le();
let payload = bytes.slice(8..8 + length as usize);
match code {
- CREATE_STREAM_CODE => Ok(EntryCommand::CreateStream(CreateStream::from_bytes(
+ CREATE_STREAM_CODE => Ok(EntryCommand::CreateStream(CreateStreamWithId::from_bytes(
payload,
)?)),
UPDATE_STREAM_CODE => Ok(EntryCommand::UpdateStream(UpdateStream::from_bytes(
@@ -105,7 +104,9 @@ impl BytesSerializable for EntryCommand {
payload,
)?)),
PURGE_STREAM_CODE => Ok(EntryCommand::PurgeStream(PurgeStream::from_bytes(payload)?)),
- CREATE_TOPIC_CODE => Ok(EntryCommand::CreateTopic(CreateTopic::from_bytes(payload)?)),
+ CREATE_TOPIC_CODE => Ok(EntryCommand::CreateTopic(CreateTopicWithId::from_bytes(
+ payload,
+ )?)),
UPDATE_TOPIC_CODE => Ok(EntryCommand::UpdateTopic(UpdateTopic::from_bytes(payload)?)),
DELETE_TOPIC_CODE => Ok(EntryCommand::DeleteTopic(DeleteTopic::from_bytes(payload)?)),
PURGE_TOPIC_CODE => Ok(EntryCommand::PurgeTopic(PurgeTopic::from_bytes(payload)?)),
@@ -116,12 +117,14 @@ impl BytesSerializable for EntryCommand {
DeletePartitions::from_bytes(payload)?,
)),
CREATE_CONSUMER_GROUP_CODE => Ok(EntryCommand::CreateConsumerGroup(
- CreateConsumerGroup::from_bytes(payload)?,
+ CreateConsumerGroupWithId::from_bytes(payload)?,
)),
DELETE_CONSUMER_GROUP_CODE => Ok(EntryCommand::DeleteConsumerGroup(
DeleteConsumerGroup::from_bytes(payload)?,
)),
- CREATE_USER_CODE => Ok(EntryCommand::CreateUser(CreateUser::from_bytes(payload)?)),
+ CREATE_USER_CODE => Ok(EntryCommand::CreateUser(CreateUserWithId::from_bytes(
+ payload,
+ )?)),
UPDATE_USER_CODE => Ok(EntryCommand::UpdateUser(UpdateUser::from_bytes(payload)?)),
DELETE_USER_CODE => Ok(EntryCommand::DeleteUser(DeleteUser::from_bytes(payload)?)),
CHANGE_PASSWORD_CODE => Ok(EntryCommand::ChangePassword(ChangePassword::from_bytes(
diff --git a/server/src/state/models.rs b/server/src/state/models.rs
index fa7b934b..9d6f6682 100644
--- a/server/src/state/models.rs
+++ b/server/src/state/models.rs
@@ -3,18 +3,94 @@ use bytes::{BufMut, Bytes, BytesMut};
use error_set::ErrContext;
use iggy::bytes_serializable::BytesSerializable;
use iggy::command::Command;
+use iggy::consumer_groups::create_consumer_group::CreateConsumerGroup;
use iggy::error::IggyError;
use iggy::personal_access_tokens::create_personal_access_token::CreatePersonalAccessToken;
+use iggy::streams::create_stream::CreateStream;
+use iggy::topics::create_topic::CreateTopic;
+use iggy::users::create_user::CreateUser;
use iggy::validatable::Validatable;
use serde::{Deserialize, Serialize};
use std::fmt;
use std::fmt::{Display, Formatter};
use std::str::from_utf8;
+#[derive(Debug, PartialEq, Serialize, Deserialize)]
+pub struct CreateStreamWithId {
+ pub stream_id: u32,
+ pub command: CreateStream,
+}
+
+#[derive(Debug, PartialEq, Serialize, Deserialize)]
+pub struct CreateTopicWithId {
+ pub topic_id: u32,
+ pub command: CreateTopic,
+}
+
+#[derive(Debug, PartialEq, Serialize, Deserialize)]
+pub struct CreateConsumerGroupWithId {
+ pub group_id: u32,
+ pub command: CreateConsumerGroup,
+}
+
+#[derive(Debug, PartialEq, Serialize, Deserialize)]
+pub struct CreateUserWithId {
+ pub user_id: u32,
+ pub command: CreateUser,
+}
+
#[derive(Debug, PartialEq, Serialize, Deserialize)]
pub struct CreatePersonalAccessTokenWithHash {
- pub command: CreatePersonalAccessToken,
pub hash: String,
+ pub command: CreatePersonalAccessToken,
+}
+
+impl Validatable<IggyError> for CreateStreamWithId {
+ fn validate(&self) -> Result<(), IggyError> {
+ self.command.validate()
+ }
+}
+
+impl Command for CreateStreamWithId {
+ fn code(&self) -> u32 {
+ self.command.code()
+ }
+}
+
+impl Validatable<IggyError> for CreateTopicWithId {
+ fn validate(&self) -> Result<(), IggyError> {
+ self.command.validate()
+ }
+}
+
+impl Command for CreateTopicWithId {
+ fn code(&self) -> u32 {
+ self.command.code()
+ }
+}
+
+impl Validatable<IggyError> for CreateConsumerGroupWithId {
+ fn validate(&self) -> Result<(), IggyError> {
+ self.command.validate()
+ }
+}
+
+impl Command for CreateConsumerGroupWithId {
+ fn code(&self) -> u32 {
+ self.command.code()
+ }
+}
+
+impl Validatable<IggyError> for CreateUserWithId {
+ fn validate(&self) -> Result<(), IggyError> {
+ self.command.validate()
+ }
+}
+
+impl Command for CreateUserWithId {
+ fn code(&self) -> u32 {
+ self.command.code()
+ }
}
impl Validatable<IggyError> for CreatePersonalAccessTokenWithHash {
@@ -29,14 +105,63 @@ impl Command for CreatePersonalAccessTokenWithHash {
}
}
-impl BytesSerializable for CreatePersonalAccessTokenWithHash {
+impl Display for CreateStreamWithId {
+ fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+ write!(
+ f,
+ "CreateStreamWithId {{ command: {}, stream_id: {} }}",
+ self.command, self.stream_id
+ )
+ }
+}
+
+impl Display for CreateTopicWithId {
+ fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+ write!(
+ f,
+ "CreateTopicWithId {{ command: {}, topic_id: {} }}",
+ self.command, self.topic_id
+ )
+ }
+}
+
+impl Display for CreateConsumerGroupWithId {
+ fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+ write!(
+ f,
+ "CreateConsumerGroupWithId {{ command: {}, group_id: {} }}",
+ self.command, self.group_id
+ )
+ }
+}
+
+impl Display for CreateUserWithId {
+ fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+ write!(
+ f,
+ "CreateUserWithId {{ command: {}, user_id: {} }}",
+ self.command, self.user_id
+ )
+ }
+}
+
+impl Display for CreatePersonalAccessTokenWithHash {
+ fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+ write!(
+ f,
+ "CreatePersonalAccessTokenWithHash {{ command: {}, hash: {} }}",
+ self.command, self.hash
+ )
+ }
+}
+
+impl BytesSerializable for CreateStreamWithId {
fn to_bytes(&self) -> Bytes {
let mut bytes = BytesMut::new();
+ bytes.put_u32_le(self.stream_id);
let command_bytes = self.command.to_bytes();
bytes.put_u32_le(command_bytes.len() as u32);
bytes.put_slice(&command_bytes);
- bytes.put_u32_le(self.hash.len() as u32);
- bytes.put_slice(self.hash.as_bytes());
bytes.freeze()
}
@@ -44,46 +169,197 @@ impl BytesSerializable for CreatePersonalAccessTokenWithHash {
where
Self: Sized,
{
- let mut position = 0;
+ let stream_id = u32::from_le_bytes(
+ bytes[0..4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse stream ID")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ let mut position = 4;
let command_length = u32::from_le_bytes(
bytes[position..position + 4]
.try_into()
.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to parse personal access token command length")
+ format!("{COMPONENT} (error: {error}) - failed to parse stream command length")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ position += 4;
+ let command_bytes = bytes.slice(position..position + command_length as usize);
+ let command = CreateStream::from_bytes(command_bytes).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse stream command")
+ })?;
+ Ok(Self { stream_id, command })
+ }
+}
+
+impl BytesSerializable for CreateTopicWithId {
+ fn to_bytes(&self) -> Bytes {
+ let mut bytes = BytesMut::new();
+ bytes.put_u32_le(self.topic_id);
+ let command_bytes = self.command.to_bytes();
+ bytes.put_u32_le(command_bytes.len() as u32);
+ bytes.put_slice(&command_bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<Self, IggyError>
+ where
+ Self: Sized,
+ {
+ let topic_id = u32::from_le_bytes(
+ bytes[0..4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse topic ID")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ let mut position = 4;
+ let command_length = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse topic command length")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ position += 4;
+ let command_bytes = bytes.slice(position..position + command_length as usize);
+ let command = CreateTopic::from_bytes(command_bytes).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse topic command")
+ })?;
+ Ok(Self { topic_id, command })
+ }
+}
+
+impl BytesSerializable for CreateConsumerGroupWithId {
+ fn to_bytes(&self) -> Bytes {
+ let mut bytes = BytesMut::new();
+ bytes.put_u32_le(self.group_id);
+ let command_bytes = self.command.to_bytes();
+ bytes.put_u32_le(command_bytes.len() as u32);
+ bytes.put_slice(&command_bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<Self, IggyError>
+ where
+ Self: Sized,
+ {
+ let group_id = u32::from_le_bytes(
+ bytes[0..4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse consumer group ID")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ let mut position = 4;
+ let command_length = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse consumer group command length")
})
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
position += 4;
let command_bytes = bytes.slice(position..position + command_length as usize);
- position += command_length as usize;
let command =
- CreatePersonalAccessToken::from_bytes(command_bytes).with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to parse personal access token command"
- )
+ CreateConsumerGroup::from_bytes(command_bytes).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse consumer group command")
})?;
- let hash_length = u32::from_le_bytes(
+ Ok(Self { group_id, command })
+ }
+}
+
+impl BytesSerializable for CreateUserWithId {
+ fn to_bytes(&self) -> Bytes {
+ let mut bytes = BytesMut::new();
+ bytes.put_u32_le(self.user_id);
+ let command_bytes = self.command.to_bytes();
+ bytes.put_u32_le(command_bytes.len() as u32);
+ bytes.put_slice(&command_bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<Self, IggyError>
+ where
+ Self: Sized,
+ {
+ let user_id = u32::from_le_bytes(
+ bytes[0..4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse user ID")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ let mut position = 4;
+ let command_length = u32::from_le_bytes(
bytes[position..position + 4]
.try_into()
.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to parse personal access token hash length")
+ format!("{COMPONENT} (error: {error}) - failed to parse user command length")
})
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
position += 4;
- let hash = from_utf8(&bytes[position..position + hash_length as usize])
- .map_err(|_| IggyError::InvalidUtf8)?
- .to_string();
- Ok(Self { command, hash })
+ let command_bytes = bytes.slice(position..position + command_length as usize);
+ let command = CreateUser::from_bytes(command_bytes).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse user command")
+ })?;
+ Ok(Self { user_id, command })
}
}
-impl Display for CreatePersonalAccessTokenWithHash {
- fn fmt(&self, f: &mut Formatter) -> fmt::Result {
- write!(
- f,
- "CreatePersonalAccessTokenWithHash {{ command: {}, hash: {} }}",
- self.command, self.hash
- )
+impl BytesSerializable for CreatePersonalAccessTokenWithHash {
+ fn to_bytes(&self) -> Bytes {
+ let mut bytes = BytesMut::new();
+ bytes.put_u32_le(self.hash.len() as u32);
+ bytes.put_slice(self.hash.as_bytes());
+ let command_bytes = self.command.to_bytes();
+ bytes.put_u32_le(command_bytes.len() as u32);
+ bytes.put_slice(&command_bytes);
+ bytes.freeze()
+ }
+
+ fn from_bytes(bytes: Bytes) -> Result<Self, IggyError>
+ where
+ Self: Sized,
+ {
+ let hash_length = u32::from_le_bytes(
+ bytes[0..4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse hash length")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ let hash = from_utf8(&bytes[4..4 + hash_length as usize])
+ .map_err(|_| IggyError::InvalidUtf8)?
+ .to_string();
+
+ let mut position = 4 + hash_length as usize;
+ let command_length = u32::from_le_bytes(
+ bytes[position..position + 4]
+ .try_into()
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to parse personal access token command length")
+ })
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ );
+ position += 4;
+ let command_bytes = bytes.slice(position..position + command_length as usize);
+ let command =
+ CreatePersonalAccessToken::from_bytes(command_bytes).with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to parse personal access token command"
+ )
+ })?;
+ Ok(Self { hash, command })
}
}
diff --git a/server/src/state/system.rs b/server/src/state/system.rs
index 5ba73f54..4487954c 100644
--- a/server/src/state/system.rs
+++ b/server/src/state/system.rs
@@ -11,7 +11,7 @@ use iggy::utils::expiry::IggyExpiry;
use iggy::utils::timestamp::IggyTimestamp;
use iggy::utils::topic_size::MaxTopicSize;
use std::fmt::Display;
-use tracing::debug;
+use tracing::{debug, info};
#[derive(Debug)]
pub struct SystemState {
@@ -25,7 +25,6 @@ pub struct StreamState {
pub name: String,
pub created_at: IggyTimestamp,
pub topics: AHashMap<u32, TopicState>,
- pub current_topic_id: u32,
}
#[derive(Debug)]
@@ -39,7 +38,6 @@ pub struct TopicState {
pub max_topic_size: MaxTopicSize,
pub replication_factor: Option<u8>,
pub created_at: IggyTimestamp,
- pub current_consumer_group_id: u32,
}
#[derive(Debug)]
@@ -61,6 +59,7 @@ pub struct UserState {
pub username: String,
pub password_hash: String,
pub status: UserStatus,
+ pub created_at: IggyTimestamp,
pub permissions: Option<Permissions>,
pub personal_access_tokens: AHashMap<String, PersonalAccessTokenState>,
}
@@ -75,23 +74,19 @@ impl SystemState {
pub async fn init(entries: Vec<StateEntry>) -> Result<Self, IggyError> {
let mut streams = AHashMap::new();
let mut users = AHashMap::new();
- let mut current_stream_id = 0;
- let mut current_user_id = 0;
for entry in entries {
debug!("Processing state entry: {entry}",);
match entry.command().with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to retrieve state entry command: {entry}")
})? {
EntryCommand::CreateStream(command) => {
- let stream_id = command.stream_id.unwrap_or_else(|| {
- current_stream_id += 1;
- current_stream_id
- });
+ info!("Creating stream: {command:?}");
+ let stream_id = command.stream_id;
+ let command = command.command;
let stream = StreamState {
id: stream_id,
name: command.name.clone(),
topics: AHashMap::new(),
- current_topic_id: 0,
created_at: entry.timestamp,
};
streams.insert(stream.id, stream);
@@ -115,19 +110,16 @@ impl SystemState {
// It only affects the segments which are not part of the state
}
EntryCommand::CreateTopic(command) => {
- let stream_id = find_stream_id(&streams, &command.stream_id);
+ let stream_id = find_stream_id(&streams, &command.command.stream_id);
let stream = streams
.get_mut(&stream_id)
.unwrap_or_else(|| panic!("{}", format!("Stream: {stream_id} not found")));
- let topic_id = command.topic_id.unwrap_or_else(|| {
- stream.current_topic_id += 1;
- stream.current_topic_id
- });
+ let topic_id = command.topic_id;
+ let command = command.command;
let topic = TopicState {
id: topic_id,
name: command.name,
consumer_groups: AHashMap::new(),
- current_consumer_group_id: 0,
compression_algorithm: command.compression_algorithm,
message_expiry: command.message_expiry,
max_topic_size: command.max_topic_size,
@@ -242,6 +234,8 @@ impl SystemState {
}
}
EntryCommand::CreateConsumerGroup(command) => {
+ let consumer_group_id = command.group_id;
+ let command = command.command;
let stream_id = find_stream_id(&streams, &command.stream_id);
let stream = streams
.get_mut(&stream_id)
@@ -251,10 +245,6 @@ impl SystemState {
.topics
.get_mut(&topic_id)
.unwrap_or_else(|| panic!("{}", format!("Topic: {topic_id} not found")));
- let consumer_group_id = command.group_id.unwrap_or_else(|| {
- topic.current_consumer_group_id += 1;
- topic.current_consumer_group_id
- });
let consumer_group = ConsumerGroupState {
id: consumer_group_id,
name: command.name,
@@ -278,12 +268,14 @@ impl SystemState {
topic.consumer_groups.remove(&consumer_group_id);
}
EntryCommand::CreateUser(command) => {
- current_user_id += 1;
+ let user_id = command.user_id;
+ let command = command.command;
let user = UserState {
- id: current_user_id,
+ id: user_id,
username: command.username,
password_hash: command.password, // This is already hashed
status: command.status,
+ created_at: entry.timestamp,
permissions: command.permissions,
personal_access_tokens: AHashMap::new(),
};
diff --git a/server/src/streaming/systems/users.rs b/server/src/streaming/systems/users.rs
index a732c1a0..08ce3c46 100644
--- a/server/src/streaming/systems/users.rs
+++ b/server/src/streaming/systems/users.rs
@@ -1,4 +1,5 @@
use crate::state::command::EntryCommand;
+use crate::state::models::CreateUserWithId;
use crate::state::system::UserState;
use crate::streaming::personal_access_tokens::personal_access_token::PersonalAccessToken;
use crate::streaming::session::Session;
@@ -35,7 +36,10 @@ impl System {
permissions: root.permissions.clone(),
};
self.state
- .apply(0, EntryCommand::CreateUser(command))
+ .apply(0, EntryCommand::CreateUser(CreateUserWithId {
+ user_id: root.id,
+ command
+ }))
.await
.with_error_context(|error| {
format!(