You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/06/21 11:51:33 UTC
(iggy) branch io_uring_tpc updated: fix state initialization with gate
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new b1bb34ce fix state initialization with gate
b1bb34ce is described below
commit b1bb34cede8aea7ce31f994b7f03e0d1fbdced63
Author: numinex <g....@gmail.com>
AuthorDate: Sat Jun 21 13:51:27 2025 +0200
fix state initialization with gate
---
core/server/src/bootstrap.rs | 48 +++------
core/server/src/lib.rs | 8 +-
core/server/src/main.rs | 138 +++++++++++++++++++++-----
core/server/src/shard/builder.rs | 62 +++++-------
core/server/src/shard/mod.rs | 124 +++++++++++++++++++++--
core/server/src/shard/transmission/frame.rs | 17 ++++
core/server/src/shard/transmission/message.rs | 17 ++++
core/server/src/shard/transmission/mod.rs | 18 ++++
core/server/src/state/file.rs | 4 +-
core/server/src/streaming/systems/info.rs | 55 ----------
core/server/src/streaming/systems/system.rs | 11 +-
core/server/src/streaming/utils/gate.rs | 42 ++++++++
core/server/src/streaming/utils/mod.rs | 1 +
13 files changed, 373 insertions(+), 172 deletions(-)
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index c6108568..4f356a9b 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -12,10 +12,13 @@ use crate::{
IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV,
configs::{config_provider::ConfigProviderKind, server::ServerConfig, system::SystemConfig},
server_error::ServerError,
- shard::{transmission::connector::ShardConnector, transmission::frame::ShardFrame},
- streaming::users::user::User,
+ shard::transmission::{connector::ShardConnector, frame::ShardFrame},
+ streaming::{
+ persistence::persister::{FilePersister, FileWithSyncPersister, PersisterKind},
+ users::user::User,
+ },
};
-use std::{env, fs::remove_dir_all, ops::Range, path::Path};
+use std::{env, fs::remove_dir_all, ops::Range, path::Path, sync::Arc};
pub fn create_shard_connections(shards_set: Range<usize>) -> Vec<ShardConnector<ShardFrame>> {
let shards_count = shards_set.len();
@@ -68,37 +71,7 @@ pub async fn create_directories(config: &SystemConfig) -> Result<(), IggyError>
// TODO: Move this to individual shard level
/*
- let state_entries = self.state.init().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to initialize state entries")
- })?;
- let system_state = SystemState::init(state_entries)
- .await
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to initialize system state")
- })?;
- let now = Instant::now();
- self.load_version().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to load version")
- })?;
- self.load_users(system_state.users.into_values().collect())
- .await
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to load users")
- })?;
- self.load_streams(system_state.streams.into_values().collect())
- .await
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to load streams")
- })?;
- if let Some(archiver) = self.archiver.as_ref() {
- archiver
- .init()
- .await
- .expect("Failed to initialize archiver");
- }
- info!("Initialized system in {} ms.", now.elapsed().as_millis());
- Ok(())
- */
+ */
}
pub fn create_root_user() -> User {
@@ -158,3 +131,10 @@ pub fn create_shard_executor() -> Runtime<TimeDriver<monoio::IoUringDriver>> {
let rt = Buildable::build(builder).expect("Failed to create default runtime");
rt
}
+
+pub fn resolve_persister(enforce_fsync: bool) -> Arc<PersisterKind> {
+ match enforce_fsync {
+ true => Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister)),
+ false => Arc::new(PersisterKind::File(FilePersister)),
+ }
+}
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index e64837da..3dcf64b2 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -43,11 +43,11 @@ pub mod streaming;
pub mod tcp;
pub mod versioning;
-const VERSION: &str = env!("CARGO_PKG_VERSION");
-const IGGY_ROOT_USERNAME_ENV: &str = "IGGY_ROOT_USERNAME";
-const IGGY_ROOT_PASSWORD_ENV: &str = "IGGY_ROOT_PASSWORD";
+pub const VERSION: &str = env!("CARGO_PKG_VERSION");
+pub const IGGY_ROOT_USERNAME_ENV: &str = "IGGY_ROOT_USERNAME";
+pub const IGGY_ROOT_PASSWORD_ENV: &str = "IGGY_ROOT_PASSWORD";
-pub(crate) fn map_toggle_str<'a>(enabled: bool) -> &'a str {
+pub fn map_toggle_str<'a>(enabled: bool) -> &'a str {
match enabled {
true => "enabled",
false => "disabled",
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 67728862..47a23f47 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+use std::sync::Arc;
use std::thread::available_parallelism;
use anyhow::Result;
@@ -23,33 +24,32 @@ use clap::Parser;
use dotenvy::dotenv;
use error_set::ErrContext;
use figlet_rs::FIGfont;
+use iggy_common::create_user::CreateUser;
+use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyError};
use server::args::Args;
use server::bootstrap::{
create_default_executor, create_directories, create_root_user, create_shard_connections,
- create_shard_executor, load_config,
+ create_shard_executor, load_config, resolve_persister,
};
-use server::channels::commands::archive_state::ArchiveStateExecutor;
-use server::channels::commands::clean_personal_access_tokens::CleanPersonalAccessTokensExecutor;
-use server::channels::commands::maintain_messages::MaintainMessagesExecutor;
-use server::channels::commands::print_sysinfo::SysInfoPrintExecutor;
-use server::channels::commands::save_messages::SaveMessagesExecutor;
-use server::channels::commands::verify_heartbeats::VerifyHeartbeatsExecutor;
-use server::channels::handler::BackgroundServerCommandHandler;
-use server::configs::config_provider::{self, ConfigProviderKind};
-use server::configs::server::ServerConfig;
-use server::http::http_server;
+use server::configs::config_provider::{self};
#[cfg(not(feature = "tokio-console"))]
use server::log::logger::Logging;
#[cfg(feature = "tokio-console")]
use server::log::tokio_console::Logging;
-use server::quic::quic_server;
use server::server_error::ServerError;
use server::shard::IggyShard;
-use server::streaming::systems::system::{SharedSystem, System};
-use server::streaming::utils::MemoryPool;
-use server::tcp::tcp_server;
+use server::state::StateKind;
+use server::state::command::EntryCommand;
+use server::state::file::FileState;
+use server::state::models::CreateUserWithId;
+use server::state::system::SystemState;
+use server::streaming::utils::gate::Gate;
+use server::versioning::SemanticVersion;
+use server::{IGGY_ROOT_USERNAME_ENV, map_toggle_str};
use tokio::time::Instant;
-use tracing::{info, instrument};
+use tracing::{error, info, instrument};
+
+const COMPONENT: &str = "MAIN";
#[instrument(skip_all, name = "trace_start_server")]
fn main() -> Result<(), ServerError> {
@@ -88,7 +88,11 @@ fn main() -> Result<(), ServerError> {
config
})?;
- // Create directories and root user.
+ // Initialize logging
+ let mut logging = Logging::new(config.telemetry.clone());
+ logging.early_init();
+
+ // Create directories.
// Remove `local_data` directory if run with `--fresh` flag.
std::thread::scope(|scope| {
scope
@@ -117,13 +121,9 @@ fn main() -> Result<(), ServerError> {
})
})
.join()
- .expect("Failed to create directories and root user")
+ .expect("Failed join thread")
})
- .with_error_context(|err| format!("Failed to create directories, err: {err}"))?;
-
- // Initialize logging
- let mut logging = Logging::new(config.telemetry.clone());
- logging.early_init();
+ .with_error_context(|err| format!("Failed to init server: {err}"))?;
// TODO: Make this configurable from config as a range
// for example this instance of Iggy will use cores from 0..4
@@ -132,9 +132,11 @@ fn main() -> Result<(), ServerError> {
let shards_set = 0..shards_count;
let connections = create_shard_connections(shards_set.clone());
for shard_id in shards_set {
+ let gate: Arc<Gate<()>> = Arc::new(Gate::new());
let id = shard_id as u16;
let connections = connections.clone();
- let server_config = config.clone();
+ let config = config.clone();
+ let state_persister = resolve_persister(config.system.state.enforce_fsync);
std::thread::Builder::new()
.name(format!("shard-{id}"))
.spawn(move || {
@@ -143,11 +145,97 @@ fn main() -> Result<(), ServerError> {
let mut rt = create_shard_executor();
rt.block_on(async move {
+ let version = SemanticVersion::current().expect("Invalid version");
+ info!(
+ "Server-side encryption is {}.",
+ map_toggle_str(config.system.encryption.enabled)
+ );
+ let encryptor: Option<EncryptorKind> = match config.system.encryption.enabled {
+ true => Some(EncryptorKind::Aes256Gcm(
+ Aes256GcmEncryptor::from_base64_key(&config.system.encryption.key)
+ .unwrap(),
+ )),
+ false => None,
+ };
+
+ let state = StateKind::File(FileState::new(
+ &config.system.get_state_messages_file_path(),
+ &version,
+ state_persister,
+ encryptor.clone(),
+ ));
+
+ let gate = gate.clone();
+ gate.with_async::<Result<(), IggyError>>(async |gate_state| {
+ if let Some(_) = gate_state.inner() {
+ return Ok(());
+ }
+
+ let state_entries = state.load_entries().await.with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to load state entries"
+ )
+ })?;
+ let root_exists = state_entries
+ .iter()
+ .find(|entry| {
+ entry
+ .command()
+ .and_then(|command| match command {
+ EntryCommand::CreateUser(payload)
+ if payload.command.username
+ == IGGY_ROOT_USERNAME_ENV =>
+ {
+ Ok(true)
+ }
+ _ => Ok(false),
+ })
+ .map_or_else(
+ |err| {
+ error!("Failed to check if root user exists: {err}");
+ false
+ },
+ |v| v,
+ )
+ })
+ .is_some();
+
+ if !root_exists {
+ info!("No users found, creating the root user...");
+ let root = create_root_user();
+ let command = CreateUser {
+ username: root.username.clone(),
+ password: root.password.clone(),
+ status: root.status,
+ permissions: root.permissions.clone(),
+ };
+ state
+ .apply(0, &EntryCommand::CreateUser(CreateUserWithId {
+ user_id: root.id,
+ command
+ }))
+ .await
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to apply create user command, username: {}",
+ root.username
+ )
+ })?;
+ }
+
+ gate_state.set_result(());
+ Ok(())
+ })
+ .await;
+
let builder = IggyShard::builder();
let mut shard = builder
.id(id)
.connections(connections)
- .server_config(server_config)
+ .config(config)
+ .encryptor(encryptor)
+ .version(version)
+ .state(state)
.build()
.await;
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 36fe0e54..e796bd11 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -22,14 +22,12 @@ use iggy_common::{Aes256GcmEncryptor, EncryptorKind};
use tracing::info;
use crate::{
+ bootstrap::resolve_persister,
configs::server::ServerConfig,
map_toggle_str,
shard::Shard,
state::{StateKind, file::FileState},
- streaming::{
- persistence::persister::{FilePersister, FileWithSyncPersister, PersisterKind},
- storage::SystemStorage,
- },
+ streaming::storage::SystemStorage,
versioning::SemanticVersion,
};
@@ -40,6 +38,9 @@ pub struct IggyShardBuilder {
id: Option<u16>,
connections: Option<Vec<ShardConnector<ShardFrame>>>,
config: Option<ServerConfig>,
+ encryptor: Option<EncryptorKind>,
+ version: Option<SemanticVersion>,
+ state: Option<StateKind>,
}
impl IggyShardBuilder {
@@ -53,16 +54,32 @@ impl IggyShardBuilder {
self
}
- pub fn server_config(mut self, config: ServerConfig) -> Self {
+ pub fn config(mut self, config: ServerConfig) -> Self {
self.config = Some(config);
self
}
+ pub fn encryptor(mut self, encryptor: Option<EncryptorKind>) -> Self {
+ self.encryptor = encryptor;
+ self
+ }
+
+ pub fn version(mut self, version: SemanticVersion) -> Self {
+ self.version = Some(version);
+ self
+ }
+
+ pub fn state(mut self, state: StateKind) -> Self {
+ self.state = Some(state);
+ self
+ }
+
// TODO: Too much happens in there, some of those bootstrapping logic should be moved outside.
pub async fn build(self) -> IggyShard {
let id = self.id.unwrap();
let config = self.config.unwrap();
let connections = self.connections.unwrap();
+ let state = self.state.unwrap();
let (stop_sender, stop_receiver, frame_receiver) = connections
.iter()
.filter(|c| c.id == id)
@@ -76,33 +93,7 @@ impl IggyShardBuilder {
.next()
.expect("Failed to find connection with the specified ID");
let shards = connections.into_iter().map(Shard::new).collect();
-
- //TODO: This can be discrete step in the builder bootstrapped from main function.
- let version = SemanticVersion::current().expect("Invalid version");
-
- //TODO: This can be discrete step in the builder bootstrapped from main function.
- info!(
- "Server-side encryption is {}.",
- map_toggle_str(config.system.encryption.enabled)
- );
- let encryptor: Option<Arc<EncryptorKind>> = match config.system.encryption.enabled {
- true => Some(Arc::new(EncryptorKind::Aes256Gcm(
- Aes256GcmEncryptor::from_base64_key(&config.system.encryption.key).unwrap(),
- ))),
- false => None,
- };
-
- //TODO: This can be discrete step in the builder bootstrapped from main function.
- let state_persister = Self::resolve_persister(config.system.state.enforce_fsync);
- let state = Rc::new(StateKind::File(FileState::new(
- &config.system.get_state_messages_file_path(),
- &version,
- state_persister,
- encryptor.clone(),
- )));
-
- //TODO: This can be discrete step in the builder bootstrapped from main function.
- let partition_persister = Self::resolve_persister(config.system.partition.enforce_fsync);
+ let partition_persister = resolve_persister(config.system.partition.enforce_fsync);
let storage = Rc::new(SystemStorage::new(
config.system.clone(),
partition_persister,
@@ -120,11 +111,4 @@ impl IggyShardBuilder {
frame_receiver: Cell::new(Some(frame_receiver)),
}
}
-
- fn resolve_persister(enforce_fsync: bool) -> Arc<PersisterKind> {
- match enforce_fsync {
- true => Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister)),
- false => Arc::new(PersisterKind::File(FilePersister)),
- }
- }
}
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 6a6d35b8..2f0929bf 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -22,11 +22,13 @@ pub mod transmission;
use ahash::HashMap;
use builder::IggyShardBuilder;
+use error_set::ErrContext;
use iggy_common::IggyError;
use namespace::IggyNamespace;
use std::{
cell::{Cell, RefCell},
rc::Rc,
+ str::FromStr,
sync::Arc,
time::Instant,
};
@@ -37,9 +39,17 @@ use crate::{
bootstrap::create_root_user,
configs::server::ServerConfig,
shard::transmission::frame::ShardFrame,
- state::{StateKind, file::FileState},
- streaming::storage::SystemStorage,
+ state::{
+ StateKind,
+ file::FileState,
+ system::{SystemState, UserState},
+ },
+ streaming::{storage::SystemStorage, systems::info::SystemInfo},
+ versioning::SemanticVersion,
};
+
+pub const COMPONENT: &str = "SHARD";
+
pub(crate) struct Shard {
id: u16,
connection: ShardConnector<ShardFrame>,
@@ -70,7 +80,7 @@ pub struct IggyShard {
// TODO: Refactor.
pub(crate) storage: Rc<SystemStorage>,
- pub(crate) state: Rc<StateKind>,
+ pub(crate) state: StateKind,
//pub(crate) encryptor: Option<Rc<dyn Encryptor>>,
config: ServerConfig,
//pub(crate) client_manager: RefCell<ClientManager>,
@@ -92,10 +102,12 @@ impl IggyShard {
//let state_entries = self.state.init().await?;
//let system_state = SystemState::init(state_entries).await?;
//let user = create_root_user();
- self.load_state().await;
- self.load_users().await;
+ self.load_version().await?;
+ let state = self.load_state().await?;
+ //self.load_users().await;
// Add default root user.
self.load_streams().await;
+
//TODO: Fix the archiver.
/*
if let Some(archiver) = self.archiver.as_ref() {
@@ -109,12 +121,106 @@ impl IggyShard {
Ok(())
}
- async fn load_state(&self) {
- todo!()
+ async fn load_version(&self) -> Result<(), IggyError> {
+ async fn update_system_info(
+ storage: &Rc<SystemStorage>,
+ system_info: &mut SystemInfo,
+ version: &SemanticVersion,
+ ) -> Result<(), IggyError> {
+ system_info.update_version(version);
+ storage.info.save(system_info).await?;
+ Ok(())
+ }
+
+ let current_version = SemanticVersion::current()?;
+ let mut system_info;
+ let load_system_info = self.storage.info.load().await;
+ if load_system_info.is_err() {
+ let error = load_system_info.err().unwrap();
+ if let IggyError::ResourceNotFound(_) = error {
+ info!("System info not found, creating...");
+ system_info = SystemInfo::default();
+ update_system_info(&self.storage, &mut system_info, ¤t_version).await?;
+ } else {
+ return Err(error);
+ }
+ } else {
+ system_info = load_system_info.unwrap();
+ }
+
+ info!("Loaded {system_info}.");
+ let loaded_version = SemanticVersion::from_str(&system_info.version.version)?;
+ if current_version.is_equal_to(&loaded_version) {
+ info!("System version {current_version} is up to date.");
+ } else if current_version.is_greater_than(&loaded_version) {
+ info!(
+ "System version {current_version} is greater than {loaded_version}, checking the available migrations..."
+ );
+ update_system_info(&self.storage, &mut system_info, ¤t_version).await?;
+ } else {
+ info!(
+ "System version {current_version} is lower than {loaded_version}, possible downgrade."
+ );
+ update_system_info(&self.storage, &mut system_info, ¤t_version).await?;
+ }
+
+ Ok(())
}
- async fn load_users(&self) {
- todo!()
+ async fn load_state(&self) -> Result<SystemState, IggyError> {
+ let root = create_root_user();
+ let state_entries = self.state.init().await.with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to initialize state entries")
+ })?;
+ let system_state = SystemState::init(state_entries)
+ .await
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to initialize system state")
+ })?;
+ Ok(system_state)
+ }
+
+ async fn load_users(&mut self, users: Vec<UserState>) -> Result<(), IggyError> {
+ info!("Loading users...");
+ /*
+
+ for user_state in users.into_iter() {
+ let mut user = User::with_password(
+ user_state.id,
+ &user_state.username,
+ user_state.password_hash,
+ user_state.status,
+ user_state.permissions,
+ );
+
+ user.created_at = user_state.created_at;
+ user.personal_access_tokens = user_state
+ .personal_access_tokens
+ .into_values()
+ .map(|token| {
+ (
+ Arc::new(token.token_hash.clone()),
+ PersonalAccessToken::raw(
+ user_state.id,
+ &token.name,
+ &token.token_hash,
+ token.expiry_at,
+ ),
+ )
+ })
+ .collect();
+ self.users.insert(user_state.id, user);
+ }
+
+ let users_count = self.users.len();
+ let current_user_id = self.users.keys().max().unwrap_or(&1);
+ USER_ID.store(current_user_id + 1, Ordering::SeqCst);
+ self.permissioner
+ .init(&self.users.values().collect::<Vec<&User>>());
+ self.metrics.increment_users(users_count as u32);
+ info!("Initialized {users_count} user(s).");
+ */
+ Ok(())
}
async fn load_streams(&self) {
diff --git a/core/server/src/shard/transmission/frame.rs b/core/server/src/shard/transmission/frame.rs
index ac60863d..ec63daf0 100644
--- a/core/server/src/shard/transmission/frame.rs
+++ b/core/server/src/shard/transmission/frame.rs
@@ -1,3 +1,20 @@
+/* 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 async_channel::Sender;
use bytes::Bytes;
use iggy_common::IggyError;
diff --git a/core/server/src/shard/transmission/message.rs b/core/server/src/shard/transmission/message.rs
index e9f1dd4b..3572c0ec 100644
--- a/core/server/src/shard/transmission/message.rs
+++ b/core/server/src/shard/transmission/message.rs
@@ -1,3 +1,20 @@
+/* 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::binary::command::ServerCommand;
#[derive(Debug)]
diff --git a/core/server/src/shard/transmission/mod.rs b/core/server/src/shard/transmission/mod.rs
index b871152a..1f9a79ed 100644
--- a/core/server/src/shard/transmission/mod.rs
+++ b/core/server/src/shard/transmission/mod.rs
@@ -1,3 +1,21 @@
+/* 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.
+ */
+
pub mod connector;
pub mod frame;
pub mod message;
diff --git a/core/server/src/state/file.rs b/core/server/src/state/file.rs
index f162ed75..aa14b501 100644
--- a/core/server/src/state/file.rs
+++ b/core/server/src/state/file.rs
@@ -47,7 +47,7 @@ pub struct FileState {
version: u32,
path: String,
persister: Arc<PersisterKind>,
- encryptor: Option<Arc<EncryptorKind>>,
+ encryptor: Option<EncryptorKind>,
}
impl FileState {
@@ -55,7 +55,7 @@ impl FileState {
path: &str,
version: &SemanticVersion,
persister: Arc<PersisterKind>,
- encryptor: Option<Arc<EncryptorKind>>,
+ encryptor: Option<EncryptorKind>,
) -> Self {
Self {
current_index: AtomicU64::new(0),
diff --git a/core/server/src/streaming/systems/info.rs b/core/server/src/streaming/systems/info.rs
index 61c1bcbb..3c20fd63 100644
--- a/core/server/src/streaming/systems/info.rs
+++ b/core/server/src/streaming/systems/info.rs
@@ -16,15 +16,11 @@
* under the License.
*/
-use crate::streaming::systems::system::System;
use crate::versioning::SemanticVersion;
-use iggy_common::IggyError;
use serde::{Deserialize, Serialize};
use std::collections::hash_map::DefaultHasher;
use std::fmt::Display;
use std::hash::{Hash, Hasher};
-use std::str::FromStr;
-use tracing::info;
#[derive(Debug, Serialize, Deserialize, Default)]
pub struct SystemInfo {
@@ -46,57 +42,6 @@ pub struct Migration {
pub applied_at: u64,
}
-impl System {
- pub(crate) async fn load_version(&mut self) -> Result<(), IggyError> {
- let current_version = SemanticVersion::current()?;
- let mut system_info;
- let load_system_info = self.storage.info.load().await;
- if load_system_info.is_err() {
- let error = load_system_info.err().unwrap();
- if let IggyError::ResourceNotFound(_) = error {
- info!("System info not found, creating...");
- system_info = SystemInfo::default();
- self.update_system_info(&mut system_info, ¤t_version)
- .await?;
- } else {
- return Err(error);
- }
- } else {
- system_info = load_system_info.unwrap();
- }
-
- info!("Loaded {system_info}.");
- let loaded_version = SemanticVersion::from_str(&system_info.version.version)?;
- if current_version.is_equal_to(&loaded_version) {
- info!("System version {current_version} is up to date.");
- } else if current_version.is_greater_than(&loaded_version) {
- info!(
- "System version {current_version} is greater than {loaded_version}, checking the available migrations..."
- );
- self.update_system_info(&mut system_info, ¤t_version)
- .await?;
- } else {
- info!(
- "System version {current_version} is lower than {loaded_version}, possible downgrade."
- );
- self.update_system_info(&mut system_info, ¤t_version)
- .await?;
- }
-
- Ok(())
- }
-
- async fn update_system_info(
- &self,
- system_info: &mut SystemInfo,
- version: &SemanticVersion,
- ) -> Result<(), IggyError> {
- system_info.update_version(version);
- self.storage.info.save(system_info).await?;
- Ok(())
- }
-}
-
impl SystemInfo {
pub fn update_version(&mut self, version: &SemanticVersion) {
self.version.version = version.to_string();
diff --git a/core/server/src/streaming/systems/system.rs b/core/server/src/streaming/systems/system.rs
index 761f6d1b..b3eb3804 100644
--- a/core/server/src/streaming/systems/system.rs
+++ b/core/server/src/streaming/systems/system.rs
@@ -102,10 +102,10 @@ impl System {
map_toggle_str(config.encryption.enabled)
);
- let encryptor: Option<Arc<EncryptorKind>> = match config.encryption.enabled {
- true => Some(Arc::new(EncryptorKind::Aes256Gcm(
+ let encryptor: Option<EncryptorKind> = match config.encryption.enabled {
+ true => Some(EncryptorKind::Aes256Gcm(
Aes256GcmEncryptor::from_base64_key(&config.encryption.key).unwrap(),
- ))),
+ )),
false => None,
};
@@ -122,7 +122,7 @@ impl System {
config.clone(),
SystemStorage::new(config, partition_persister),
state,
- encryptor,
+ encryptor.map(Arc::new),
data_maintenance_config,
pat_config,
)
@@ -224,9 +224,12 @@ impl System {
format!("{COMPONENT} (error: {error}) - failed to initialize system state")
})?;
let now = Instant::now();
+ //DONE
+ /*
self.load_version().await.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to load version")
})?;
+ */
self.load_users(system_state.users.into_values().collect())
.await
.with_error_context(|error| {
diff --git a/core/server/src/streaming/utils/gate.rs b/core/server/src/streaming/utils/gate.rs
new file mode 100644
index 00000000..315250fa
--- /dev/null
+++ b/core/server/src/streaming/utils/gate.rs
@@ -0,0 +1,42 @@
+use std::sync::{Condvar, Mutex};
+
+#[derive(Default)]
+pub struct Gate<T> {
+ state: Mutex<GateState<T>>,
+}
+
+#[derive(Default)]
+pub struct GateState<T> {
+ result: Option<T>,
+}
+
+impl<T> GateState<T> {
+ pub fn set_result(&mut self, result: T) {
+ self.result = Some(result);
+ }
+
+ pub fn inner(&self) -> &Option<T> {
+ &self.result
+ }
+}
+
+impl<T> Gate<T>
+where
+ T: Default,
+{
+ pub fn new() -> Self {
+ Gate {
+ state: Default::default(),
+ }
+ }
+
+ pub async fn with_async<R>(&self, f: impl AsyncFnOnce(&mut GateState<T>) -> R) {
+ let mut guard = self.state.lock().unwrap();
+ f(&mut guard).await;
+ }
+
+ pub async fn with_once<R>(&self, f: impl FnOnce(&mut GateState<T>) -> R) {
+ let mut guard = self.state.lock().unwrap();
+ f(&mut guard);
+ }
+}
diff --git a/core/server/src/streaming/utils/mod.rs b/core/server/src/streaming/utils/mod.rs
index e441a051..1f1c731f 100644
--- a/core/server/src/streaming/utils/mod.rs
+++ b/core/server/src/streaming/utils/mod.rs
@@ -18,6 +18,7 @@
pub mod crypto;
pub mod file;
+pub mod gate;
pub mod hash;
pub mod head_tail_buf;
pub mod random_id;