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

Use sqlite pool guard

Browse Source
This commit is contained in:
Andrej Mihajlov
2025-06-02 17:36:24 +02:00
committed by Bogdan-Ștefan Neacşu
parent a70a408098
commit e6387e080f
10 changed files with 91 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+4 -2
View File
@@ -5265,6 +5265,7 @@ dependencies = [
"nym-sphinx",
"nym-task",
"sqlx",
"sqlx-pool-guard",
"thiserror 2.0.12",
"time",
"tokio",
@@ -5465,6 +5466,7 @@ dependencies = [
"nym-ecash-time",
"serde",
"sqlx",
"sqlx-pool-guard",
"thiserror 2.0.12",
"tokio",
"zeroize",
@@ -7936,9 +7938,9 @@ dependencies = [
[[package]]
name = "proc_pidinfo"
version = "0.1.2"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1724fdb3f7349e4c58806dfb29dbc3b5e2f8b0997bb5117acf8db6e555867a91"
checksum = "af53dad2390f8df98dda1e4188322bdf2f91c86cf6001f51d10d64451edf463a"
dependencies = [
"libc",
]
Cargo.toml
+3 -2
View File
@@ -122,7 +122,8 @@ members = [
"service-providers/authenticator",
"service-providers/common",
"service-providers/ip-packet-router",
"service-providers/network-requester", "sqlx-pool-guard",
"service-providers/network-requester",
"sqlx-pool-guard",
"tools/echo-server",
"tools/internal/contract-state-importer/importer-cli",
"tools/internal/contract-state-importer/importer-contract",
@@ -283,7 +284,7 @@ petgraph = "0.6.5"
pin-project = "1.1"
pin-project-lite = "0.2.16"
publicsuffix = "2.3.0"
proc_pidinfo = "0.1"
proc_pidinfo = "0.1.3"
quote = "1"
rand = "0.8.5"
rand_chacha = "0.3"
common/client-core/surb-storage/Cargo.toml
+1
View File
@@ -17,6 +17,7 @@ tokio = { workspace = true, features = ["fs"] }
nym-crypto = { path = "../../crypto", optional = true, default-features = false }
nym-sphinx = { path = "../../nymsphinx" }
nym-task = { path = "../../task" }
sqlx-pool-guard = { path = "../../../sqlx-pool-guard" }
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.sqlx]
common/client-core/surb-storage/src/backend/fs_backend/manager.rs
+30 -25
View File
@@ -15,9 +15,11 @@ use sqlx::{
};
use std::path::Path;
use sqlx_pool_guard::SqlitePoolGuard;
#[derive(Debug, Clone)]
pub struct StorageManager {
connection_pool: sqlx::SqlitePool,
connection_pool: SqlitePoolGuard,
}
// all SQL goes here
@@ -37,7 +39,7 @@ impl StorageManager {
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.synchronous(SqliteSynchronous::Normal)
.auto_vacuum(SqliteAutoVacuum::Incremental)
.filename(database_path)
.filename(&database_path)
.create_if_missing(fresh)
.disable_statement_logging();
@@ -49,8 +51,11 @@ impl StorageManager {
}
};
let connection_pool =
SqlitePoolGuard::new(database_path.as_ref().to_path_buf(), connection_pool);
if let Err(err) = sqlx::migrate!("./fs_surbs_migrations")
.run(&connection_pool)
.run(&*connection_pool)
.await
{
error!("Failed to initialize SQLx database: {err}");
@@ -70,35 +75,35 @@ impl StorageManager {
#[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'")
.fetch_optional(&self.connection_pool)
.fetch_optional(&*self.connection_pool)
.await
.map(|r| r.is_some())
}
pub async fn create_status_table(&self) -> Result<(), sqlx::Error> {
sqlx::query!("INSERT INTO status(flush_in_progress, previous_flush_timestamp, client_in_use) VALUES (0, 0, 1)")
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn get_flush_status(&self) -> Result<bool, sqlx::Error> {
sqlx::query!("SELECT flush_in_progress FROM status;")
.fetch_one(&self.connection_pool)
.fetch_one(&*self.connection_pool)
.await
.map(|r| r.flush_in_progress > 0)
}
pub async fn set_previous_flush_timestamp(&self, timestamp: i64) -> Result<(), sqlx::Error> {
sqlx::query!("UPDATE status SET previous_flush_timestamp = ?", timestamp)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn get_previous_flush_timestamp(&self) -> Result<i64, sqlx::Error> {
sqlx::query!("SELECT previous_flush_timestamp FROM status;")
.fetch_one(&self.connection_pool)
.fetch_one(&*self.connection_pool)
.await
.map(|r| r.previous_flush_timestamp)
}
@@ -106,14 +111,14 @@ impl StorageManager {
pub async fn set_flush_status(&self, in_progress: bool) -> Result<(), sqlx::Error> {
let in_progress_int = i64::from(in_progress);
sqlx::query!("UPDATE status SET flush_in_progress = ?", in_progress_int)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn get_client_in_use_status(&self) -> Result<bool, sqlx::Error> {
sqlx::query!("SELECT client_in_use FROM status;")
.fetch_one(&self.connection_pool)
.fetch_one(&*self.connection_pool)
.await
.map(|r| r.client_in_use > 0)
}
@@ -121,21 +126,21 @@ impl StorageManager {
pub async fn set_client_in_use_status(&self, in_use: bool) -> Result<(), sqlx::Error> {
let in_use_int = i64::from(in_use);
sqlx::query!("UPDATE status SET client_in_use = ?", in_use_int)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn delete_all_tags(&self) -> Result<(), sqlx::Error> {
sqlx::query!("DELETE FROM sender_tag;")
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn get_tags(&self) -> Result<Vec<StoredSenderTag>, sqlx::Error> {
sqlx::query_as!(StoredSenderTag, "SELECT * FROM sender_tag;",)
.fetch_all(&self.connection_pool)
.fetch_all(&*self.connection_pool)
.await
}
@@ -147,21 +152,21 @@ impl StorageManager {
stored_tag.recipient,
stored_tag.tag
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn delete_all_reply_keys(&self) -> Result<(), sqlx::Error> {
sqlx::query!("DELETE FROM reply_key;")
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn get_reply_keys(&self) -> Result<Vec<StoredReplyKey>, sqlx::Error> {
sqlx::query_as!(StoredReplyKey, "SELECT * FROM reply_key;",)
.fetch_all(&self.connection_pool)
.fetch_all(&*self.connection_pool)
.await
}
@@ -177,14 +182,14 @@ impl StorageManager {
stored_reply_key.reply_key,
stored_reply_key.sent_at_timestamp
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
pub async fn get_surb_senders(&self) -> Result<Vec<StoredSurbSender>, sqlx::Error> {
sqlx::query_as!(StoredSurbSender, "SELECT * FROM reply_surb_sender;",)
.fetch_all(&self.connection_pool)
.fetch_all(&*self.connection_pool)
.await
}
@@ -199,7 +204,7 @@ impl StorageManager {
stored_surb_sender.tag,
stored_surb_sender.last_sent_timestamp
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?
.last_insert_rowid();
Ok(id)
@@ -214,17 +219,17 @@ impl StorageManager {
"SELECT * FROM reply_surb WHERE reply_surb_sender_id = ?",
sender_id
)
.fetch_all(&self.connection_pool)
.fetch_all(&*self.connection_pool)
.await
}
pub async fn delete_all_reply_surb_data(&self) -> Result<(), sqlx::Error> {
sqlx::query!("DELETE FROM reply_surb;")
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
sqlx::query!("DELETE FROM reply_surb_sender;")
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
@@ -241,7 +246,7 @@ impl StorageManager {
stored_reply_surb.reply_surb_sender_id,
stored_reply_surb.reply_surb
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
@@ -255,7 +260,7 @@ impl StorageManager {
SELECT min_reply_surb_threshold as "min_reply_surb_threshold: u32", max_reply_surb_threshold as "max_reply_surb_threshold: u32" FROM reply_surb_storage_metadata;
"#,
)
.fetch_one(&self.connection_pool)
.fetch_one(&*self.connection_pool)
.await
}
@@ -269,7 +274,7 @@ impl StorageManager {
"#,
metadata.min_reply_surb_threshold,
metadata.max_reply_surb_threshold,
).execute(&self.connection_pool).await?;
).execute(&*self.connection_pool).await?;
Ok(())
}
}
common/credential-storage/Cargo.toml
+8 -2
View File
@@ -19,6 +19,7 @@ zeroize = { workspace = true, features = ["zeroize_derive"] }
nym-credentials = { path = "../credentials" }
nym-compact-ecash = { path = "../nym_offline_compact_ecash" }
nym-ecash-time = { path = "../ecash-time" }
sqlx-pool-guard = { path = "../../sqlx-pool-guard" }
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.sqlx]
@@ -31,8 +32,13 @@ features = ["rt-multi-thread", "net", "signal", "fs"]
[build-dependencies]
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
sqlx = { workspace = true, features = [
"runtime-tokio-rustls",
"sqlite",
"macros",
"migrate",
] }
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
[features]
persistent-storage = ["bincode", "serde"]
persistent-storage = ["bincode", "serde"]
common/credential-storage/src/backends/sqlite.rs
+17 -16
View File
@@ -7,10 +7,11 @@ use crate::models::{
};
use nym_ecash_time::Date;
use sqlx::{Executor, Sqlite, Transaction};
use sqlx_pool_guard::SqlitePoolGuard;
#[derive(Clone)]
pub struct SqliteEcashTicketbookManager {
connection_pool: sqlx::SqlitePool,
connection_pool: SqlitePoolGuard,
}
impl SqliteEcashTicketbookManager {
@@ -19,7 +20,7 @@ impl SqliteEcashTicketbookManager {
/// # Arguments
///
/// * `connection_pool`: database connection pool to use.
pub fn new(connection_pool: sqlx::SqlitePool) -> Self {
pub fn new(connection_pool: SqlitePoolGuard) -> Self {
SqliteEcashTicketbookManager { connection_pool }
}
@@ -33,7 +34,7 @@ impl SqliteEcashTicketbookManager {
"DELETE FROM ecash_ticketbook WHERE expiration_date <= ?",
deadline
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
@@ -60,7 +61,7 @@ impl SqliteEcashTicketbookManager {
data,
expiration_date,
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
@@ -90,7 +91,7 @@ impl SqliteEcashTicketbookManager {
epoch_id,
total_tickets,
used_tickets,
).execute(&self.connection_pool).await?;
).execute(&*self.connection_pool).await?;
Ok(())
}
@@ -105,7 +106,7 @@ impl SqliteEcashTicketbookManager {
"#,
)
.bind(data)
.fetch_optional(&self.connection_pool)
.fetch_optional(&*self.connection_pool)
.await?
.is_some();
@@ -121,7 +122,7 @@ impl SqliteEcashTicketbookManager {
FROM ecash_ticketbook
"#,
)
.fetch_all(&self.connection_pool)
.fetch_all(&*self.connection_pool)
.await
}
@@ -143,7 +144,7 @@ impl SqliteEcashTicketbookManager {
ticketbook_id,
expected_current_total_spent
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?
.rows_affected();
Ok(affected > 0)
@@ -153,7 +154,7 @@ impl SqliteEcashTicketbookManager {
&self,
) -> Result<Vec<StoredPendingTicketbook>, sqlx::Error> {
sqlx::query_as("SELECT * FROM pending_issuance")
.fetch_all(&self.connection_pool)
.fetch_all(&*self.connection_pool)
.await
}
@@ -165,7 +166,7 @@ impl SqliteEcashTicketbookManager {
"DELETE FROM pending_issuance WHERE deposit_id = ?",
pending_id
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
@@ -182,7 +183,7 @@ impl SqliteEcashTicketbookManager {
"#,
epoch_id
)
.fetch_optional(&self.connection_pool)
.fetch_optional(&*self.connection_pool)
.await
}
@@ -208,7 +209,7 @@ impl SqliteEcashTicketbookManager {
serialisation_revision,
epoch_id
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
@@ -225,7 +226,7 @@ impl SqliteEcashTicketbookManager {
"#,
epoch_id
)
.fetch_optional(&self.connection_pool)
.fetch_optional(&*self.connection_pool)
.await
}
@@ -251,7 +252,7 @@ impl SqliteEcashTicketbookManager {
serialisation_revision,
epoch_id,
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
@@ -269,7 +270,7 @@ impl SqliteEcashTicketbookManager {
"#,
expiration_date
)
.fetch_optional(&self.connection_pool)
.fetch_optional(&*self.connection_pool)
.await
}
@@ -298,7 +299,7 @@ impl SqliteEcashTicketbookManager {
serialisation_revision,
expiration_date
)
.execute(&self.connection_pool)
.execute(&*self.connection_pool)
.await?;
Ok(())
}
common/credential-storage/src/persistent_storage/mod.rs
+7 -3
View File
@@ -37,6 +37,7 @@ use sqlx::{
sqlite::{SqliteAutoVacuum, SqliteSynchronous},
ConnectOptions,
};
use sqlx_pool_guard::SqlitePoolGuard;
use std::path::Path;
use zeroize::Zeroizing;
@@ -62,7 +63,7 @@ impl PersistentStorage {
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.synchronous(SqliteSynchronous::Normal)
.auto_vacuum(SqliteAutoVacuum::Incremental)
.filename(database_path)
.filename(&database_path)
.create_if_missing(true)
.disable_statement_logging();
@@ -74,14 +75,17 @@ impl PersistentStorage {
}
};
if let Err(err) = sqlx::migrate!("./migrations").run(&connection_pool).await {
let connection_pool =
SqlitePoolGuard::new(database_path.as_ref().to_path_buf(), connection_pool);
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());
}
Ok(PersistentStorage {
storage_manager: SqliteEcashTicketbookManager::new(connection_pool.clone()),
storage_manager: SqliteEcashTicketbookManager::new(connection_pool),
})
}
}
sqlx-pool-guard/src/macos.rs → sqlx-pool-guard/src/apple.rs
+2
View File
@@ -1,6 +1,8 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use std::{io, path::Path};
use proc_pidinfo::{
ProcFDInfo, ProcFDType, VnodeFdInfoWithPath, proc_pidfdinfo_self, proc_pidinfo_list_self,
};
sqlx-pool-guard/src/lib.rs
+17 -6
View File
@@ -12,8 +12,8 @@ use std::{
#[path = "windows.rs"]
mod imp;
#[cfg(target_os = "macos")]
#[path = "macos.rs"]
#[cfg(any(target_os = "macos", target_os = "ios"))]
#[path = "apple.rs"]
mod imp;
#[cfg(any(target_os = "linux", target_os = "android"))]
@@ -21,18 +21,25 @@ mod imp;
mod imp;
/// Max number of retry attempts
const CHECK_FILES_CLOSED_MAX_ATTEMPTS: u8 = 10;
const CHECK_FILES_CLOSED_MAX_ATTEMPTS: u8 = 20;
/// Delay between file checks
const CHECK_FILES_CLOSED_RETRY_DELAY: Duration = Duration::from_millis(100);
/// `sqlx::SqlitePool` wrapper providing a workaround for the [known bug](https://github.com/launchbadge/sqlx/issues/3217).
/// In principle after requesting to close the sqlite pool, the wrapper monitors open file descriptor and polls periodically until all database files are closed.
#[derive(Debug, Clone)]
pub struct SqlitePoolGuard {
/// Path to sqlite database file.
database_path: PathBuf,
/// Inner connection pool.
connection_pool: sqlx::SqlitePool,
}
impl Deref for SqlitePoolGuard {
type Target = sqlx::SqlitePool;
fn deref(&self) -> &Self::Target {
&self.connection_pool
}
@@ -53,8 +60,12 @@ impl SqlitePoolGuard {
}
/// Close udnerlying sqlite pool and wait for files to be closed before returning.
pub async fn close_pool(&self) {
_ = self.close_pool_inner();
pub async fn close(&self) {
// Avoid waiting for db files once the pool is marked closed to ensure that we don't wait on some other sqlite pool to close the database.
if !self.connection_pool.is_closed() {
log::info!("Closing sqlite pool: {}", self.database_path.display());
_ = self.close_pool_inner();
}
}
async fn close_pool_inner(&self) -> std::io::Result<()> {
@@ -126,7 +137,7 @@ mod tests {
#[tokio::test]
async fn test_wait_close() {
let temp_dir = tempfile::tempdir().unwrap();
let database_path = temp_dir.path().join("storage.sqlite");
let database_path = temp_dir.path().join("storage.db");
let opts = sqlx::sqlite::SqliteConnectOptions::new()
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
sqlx-pool-guard/src/linux.rs
+2
View File
@@ -1,6 +1,8 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use std::{io, path::Path};
static PROC_SELF_FD_DIR: &str = "/proc/self/fd/";
/// Check if there are no open file descriptors for the given files.