GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Close sqlite pool before erroring

Browse Source
This commit is contained in:
Andrej Mihajlov
2025-05-24 15:32:27 +02:00
parent ac12455f97
commit 3ede03e1d1
10 changed files with 147 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-core/gateways-storage/src/backend/fs_backend/error.rs
+1 -4
View File
@@ -2,8 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::BadGateway;
use std::io;
use std::path::PathBuf;
use std::{io, path::PathBuf};
use thiserror::Error;
#[derive(Debug, Error)]
@@ -19,7 +18,6 @@ pub enum StorageError {
#[error("failed to perform sqlx migration: {source}")]
MigrationError {
#[source]
#[from]
source: sqlx::migrate::MigrateError,
},
@@ -32,7 +30,6 @@ pub enum StorageError {
#[error("failed to run the SQL query: {source}")]
QueryError {
#[source]
#[from]
source: sqlx::error::Error,
},
common/client-core/gateways-storage/src/backend/fs_backend/manager.rs
+6 -4
View File
@@ -47,12 +47,14 @@ impl StorageManager {
StorageError::DatabaseConnectionError { source }
})?;
sqlx::migrate!("./fs_gateways_migrations")
if let Err(err) = sqlx::migrate!("./fs_gateways_migrations")
.run(&connection_pool)
.await
.inspect_err(|err| {
error!("Failed to initialize SQLx database: {err}");
})?;
{
error!("Failed to initialize SQLx database: {err}");
connection_pool.close().await;
return Err(err)?;
}
debug!("Database migration finished!");
Ok(StorageManager { connection_pool })
common/client-core/src/client/base_client/non_wasm_helpers.rs
+72 -28
View File
@@ -1,20 +1,22 @@
// Copyright 2022-2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::client::replies::reply_storage::{
fs_backend, CombinedReplyStorage, ReplyStorageBackend,
use crate::{
client::replies::reply_storage::{fs_backend, CombinedReplyStorage, ReplyStorageBackend},
config,
config::Config,
error::ClientCoreError,
};
use crate::config;
use crate::config::Config;
use crate::error::ClientCoreError;
#[cfg(windows)]
use log::debug;
use log::{error, info, trace};
use nym_bandwidth_controller::BandwidthController;
use nym_client_core_gateways_storage::OnDiskGatewaysDetails;
use nym_credential_storage::storage::Storage as CredentialStorage;
use nym_validator_client::nyxd;
use nym_validator_client::QueryHttpRpcNyxdClient;
use std::path::Path;
use std::{fs, io};
use nym_validator_client::{nyxd, QueryHttpRpcNyxdClient};
#[cfg(windows)]
use std::time::Duration;
use std::{io, path::Path};
use time::OffsetDateTime;
use url::Url;
@@ -22,11 +24,14 @@ async fn setup_fresh_backend<P: AsRef<Path>>(
db_path: P,
surb_config: &config::ReplySurbs,
) -> Result<fs_backend::Backend, ClientCoreError> {
info!("creating fresh surb database");
info!(
"creating fresh surb database: {}",
db_path.as_ref().display()
);
let mut storage_backend = match fs_backend::Backend::init(db_path).await {
Ok(backend) => backend,
Err(err) => {
error!("failed to setup persistent storage backend for our reply needs: {err}");
error!("setup_fresh_backend: Failed to setup persistent storage backend for our reply needs: {err}");
return Err(ClientCoreError::SurbStorageError {
source: Box::new(err),
});
@@ -40,14 +45,15 @@ async fn setup_fresh_backend<P: AsRef<Path>>(
surb_config.minimum_reply_surb_storage_threshold,
surb_config.maximum_reply_surb_storage_threshold,
);
storage_backend
.init_fresh(&mem_store)
.await
.map_err(|err| ClientCoreError::SurbStorageError {
source: Box::new(err),
})?;
Ok(storage_backend)
match storage_backend.init_fresh(&mem_store).await {
Ok(()) => Ok(storage_backend),
Err(err) => {
storage_backend.shutdown().await;
Err(ClientCoreError::SurbStorageError {
source: Box::new(err),
})
}
}
}
// fn setup_inactive_backend(surb_config: &config::ReplySurbs) -> fs_backend::Backend {
@@ -58,12 +64,11 @@ async fn setup_fresh_backend<P: AsRef<Path>>(
// )
// }
fn archive_corrupted_database<P: AsRef<Path>>(db_path: P) -> io::Result<()> {
async fn archive_corrupted_database<P: AsRef<Path>>(db_path: P) -> io::Result<()> {
let db_path = db_path.as_ref();
debug_assert!(db_path.exists());
let now = OffsetDateTime::now_utc().unix_timestamp();
let suffix = format!("_{now}.corrupted");
let new_extension =
@@ -72,11 +77,51 @@ fn archive_corrupted_database<P: AsRef<Path>>(db_path: P) -> io::Result<()> {
} else {
suffix
};
let renamed = db_path.with_extension(new_extension);
let mut renamed = db_path.to_owned();
renamed.set_extension(new_extension);
rename_db_file(&db_path, &renamed).await.inspect_err(|_| {
tracing::error!(
"Failed to rename corrupt database file: {} to {}",
db_path.display(),
renamed.display()
);
})
}
fs::rename(db_path, renamed)
#[cfg(not(windows))]
async fn rename_db_file(db_path: impl AsRef<Path>, renamed: impl AsRef<Path>) -> io::Result<()> {
tokio::fs::rename(db_path, &renamed).await
}
#[cfg(windows)]
async fn rename_db_file(db_path: impl AsRef<Path>, renamed: impl AsRef<Path>) -> io::Result<()> {
// Due to bug in sqlx (https://github.com/launchbadge/sqlx/issues/3217),
// the sqlite file can be still in use after closing sqlite connection pool
// Poll for a bit until the db file is released.
// Max number of retries
const MAX_RETRY_ATTEMPTS: u32 = 10;
// Delay between retries
const WAIT_DELAY: Duration = Duration::from_millis(100);
// Error code returned when file is still in use.
const FILE_IN_USE_ERR: i32 = 32;
let mut retry_attempt = 0;
while let Err(e) = tokio::fs::rename(db_path.as_ref(), renamed.as_ref()).await {
retry_attempt += 1;
if e.raw_os_error() == Some(FILE_IN_USE_ERR) && retry_attempt < MAX_RETRY_ATTEMPTS {
debug!(
"File {} is still open. Sleep and retry",
db_path.as_ref().display()
);
tokio::time::sleep(WAIT_DELAY).await;
} else {
return Err(e);
}
}
Ok(())
}
pub async fn setup_fs_reply_surb_backend<P: AsRef<Path>>(
@@ -87,13 +132,12 @@ pub async fn setup_fs_reply_surb_backend<P: AsRef<Path>>(
// the existing one
let db_path = db_path.as_ref();
if db_path.exists() {
info!("loading existing surb database");
info!("Loading existing surb database: {}", db_path.display());
match fs_backend::Backend::try_load(db_path, surb_config.fresh_sender_tags).await {
Ok(backend) => Ok(backend),
Err(err) => {
error!("failed to setup persistent storage backend for our reply needs: {err}. We're going to create a fresh database instead. This behaviour might change in the future");
archive_corrupted_database(db_path)?;
error!("setup_fs_reply_surb_backend: Failed to setup persistent storage backend for our reply needs: {err}. We're going to create a fresh database instead. This behaviour might change in the future");
archive_corrupted_database(db_path).await?;
setup_fresh_backend(db_path, surb_config).await
}
}
common/client-core/surb-storage/Cargo.toml
+7 -1
View File
@@ -12,6 +12,7 @@ dashmap.workspace = true
log.workspace = true
thiserror.workspace = true
time.workspace = true
tokio = { workspace = true, features = ["fs"] }
nym-crypto = { path = "../../crypto", optional = true, default-features = false }
nym-sphinx = { path = "../../nymsphinx" }
@@ -25,7 +26,12 @@ optional = true
[build-dependencies]
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
sqlx = { workspace = true, features = [
"runtime-tokio-rustls",
"sqlite",
"macros",
"migrate",
] }
[features]
fs-surb-storage = ["sqlx", "nym-crypto", "nym-crypto/hashing"]
common/client-core/surb-storage/src/backend/fs_backend/error.rs
+1 -4
View File
@@ -1,8 +1,7 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use std::io;
use std::path::PathBuf;
use std::{io, path::PathBuf};
use thiserror::Error;
#[derive(Debug, Error)]
@@ -30,7 +29,6 @@ pub enum StorageError {
#[error("failed to perform sqlx migration: {source}")]
MigrationError {
#[source]
#[from]
source: sqlx::migrate::MigrateError,
},
@@ -43,7 +41,6 @@ pub enum StorageError {
#[error("failed to run the SQL query: {source}")]
QueryError {
#[source]
#[from]
source: sqlx::error::Error,
},
common/client-core/surb-storage/src/backend/fs_backend/manager.rs
+7 -1
View File
@@ -17,7 +17,7 @@ use std::path::Path;
#[derive(Debug, Clone)]
pub struct StorageManager {
pub connection_pool: sqlx::SqlitePool,
connection_pool: sqlx::SqlitePool,
}
// all SQL goes here
@@ -54,6 +54,7 @@ impl StorageManager {
.await
{
error!("Failed to initialize SQLx database: {err}");
connection_pool.close().await;
return Err(err.into());
}
@@ -61,6 +62,11 @@ impl StorageManager {
Ok(StorageManager { connection_pool })
}
/// Close connection pool waiting for all connections to be closed.
pub async fn close_pool(&self) {
self.connection_pool.close().await;
}
#[allow(dead_code)]
pub async fn status_table_exists(&self) -> Result<bool, sqlx::Error> {
sqlx::query!("SELECT name FROM sqlite_master WHERE type='table' AND name='status'")
common/client-core/surb-storage/src/backend/fs_backend/mod.rs
+50 -31
View File
@@ -1,18 +1,21 @@
// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::backend::fs_backend::manager::StorageManager;
use crate::backend::fs_backend::models::{
ReplySurbStorageMetadata, StoredReplyKey, StoredReplySurb, StoredSenderTag, StoredSurbSender,
};
use crate::surb_storage::ReceivedReplySurbs;
use crate::{
CombinedReplyStorage, ReceivedReplySurbsMap, ReplyStorageBackend, SentReplyKeys, UsedSenderTags,
backend::fs_backend::{
manager::StorageManager,
models::{
ReplySurbStorageMetadata, StoredReplyKey, StoredReplySurb, StoredSenderTag,
StoredSurbSender,
},
},
surb_storage::ReceivedReplySurbs,
CombinedReplyStorage, ReceivedReplySurbsMap, ReplyStorageBackend, SentReplyKeys,
UsedSenderTags,
};
use async_trait::async_trait;
use log::{debug, error, info, warn};
use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
use std::fs;
use std::path::{Path, PathBuf};
use time::OffsetDateTime;
@@ -41,15 +44,17 @@ impl Backend {
}
let manager = StorageManager::init(database_path, true).await?;
manager.create_status_table().await?;
let backend = Backend {
temporary_old_path: None,
database_path: owned_path,
manager,
};
Ok(backend)
match manager.create_status_table().await {
Ok(()) => Ok(Backend {
temporary_old_path: None,
database_path: owned_path,
manager,
}),
Err(err) => {
manager.close_pool().await;
Err(err.into())
}
}
}
pub async fn try_load<P: AsRef<Path>>(
@@ -64,7 +69,28 @@ impl Backend {
}
let manager = StorageManager::init(database_path, false).await?;
match Self::try_load_inner(&manager, fresh_sender_tags).await {
Ok(()) => Ok(Backend {
temporary_old_path: None,
database_path: owned_path,
manager,
}),
Err(e) => {
manager.close_pool().await;
Err(e)
}
}
}
/// Gracefully close sqlite connection pool and drop backend.
pub async fn shutdown(self) {
self.manager.close_pool().await
}
async fn try_load_inner(
manager: &StorageManager,
fresh_sender_tags: bool,
) -> Result<(), StorageError> {
// the database flush wasn't fully finished and thus the data is in inconsistent state
// (we don't really know what's properly saved or what's not)
if manager.get_flush_status().await? {
@@ -126,20 +152,11 @@ impl Backend {
manager.delete_all_tags().await?;
}
Ok(Backend {
temporary_old_path: None,
database_path: owned_path,
// manager: StorageManagerState::Storage(manager),
manager,
})
}
async fn close_pool(&mut self) {
self.manager.connection_pool.close().await;
Ok(())
}
async fn rotate(&mut self) -> Result<(), StorageError> {
self.close_pool().await;
self.manager.close_pool().await;
let new_extension = if let Some(existing_extension) =
self.database_path.extension().and_then(|ext| ext.to_str())
@@ -152,7 +169,8 @@ impl Backend {
let mut temp_old = self.database_path.clone();
temp_old.set_extension(new_extension);
fs::rename(&self.database_path, &temp_old)
tokio::fs::rename(&self.database_path, &temp_old)
.await
.map_err(|err| StorageError::DatabaseRenameError { source: err })?;
self.manager = StorageManager::init(&self.database_path, true).await?;
self.manager.create_status_table().await?;
@@ -161,9 +179,10 @@ impl Backend {
Ok(())
}
fn remove_old(&mut self) -> Result<(), StorageError> {
async fn remove_old(&mut self) -> Result<(), StorageError> {
if let Some(old_path) = self.temporary_old_path.take() {
fs::remove_file(old_path)
tokio::fs::remove_file(old_path)
.await
.map_err(|err| StorageError::DatabaseOldFileRemoveError { source: err })
} else {
warn!("the old database file doesn't seem to exist!");
@@ -335,7 +354,7 @@ impl ReplyStorageBackend for Backend {
self.dump_reply_surb_storage_metadata(surbs_ref).await?;
self.dump_reply_surbs(surbs_ref).await?;
self.remove_old()?;
self.remove_old().await?;
self.end_storage_flush().await
}
common/credential-storage/src/persistent_storage/mod.rs
+1
View File
@@ -76,6 +76,7 @@ impl PersistentStorage {
if let Err(err) = sqlx::migrate!("./migrations").run(&connection_pool).await {
error!("Failed to perform migration on the SQLx database: {err}");
connection_pool.close().await;
return Err(err.into());
}
common/gateway-stats-storage/src/lib.rs
+1
View File
@@ -59,6 +59,7 @@ impl PersistentStatsStorage {
if let Err(err) = sqlx::migrate!("./migrations").run(&connection_pool).await {
error!("Failed to perform migration on the SQLx database: {err}");
connection_pool.close().await;
return Err(err.into());
}
common/gateway-storage/src/lib.rs
+1
View File
@@ -108,6 +108,7 @@ impl GatewayStorage {
if let Err(err) = sqlx::migrate!("./migrations").run(&connection_pool).await {
error!("Failed to perform migration on the SQLx database: {err}");
connection_pool.close().await;
return Err(err.into());
}