From e6387e080f29d8e6355f31fe157970bb56b5f434 Mon Sep 17 00:00:00 2001 From: Andrej Mihajlov Date: Mon, 2 Jun 2025 17:36:24 +0200 Subject: [PATCH] Use sqlite pool guard --- Cargo.lock | 6 +- Cargo.toml | 5 +- common/client-core/surb-storage/Cargo.toml | 1 + .../src/backend/fs_backend/manager.rs | 55 ++++++++++--------- common/credential-storage/Cargo.toml | 10 +++- .../credential-storage/src/backends/sqlite.rs | 33 +++++------ .../src/persistent_storage/mod.rs | 10 +++- sqlx-pool-guard/src/{macos.rs => apple.rs} | 2 + sqlx-pool-guard/src/lib.rs | 23 ++++++-- sqlx-pool-guard/src/linux.rs | 2 + 10 files changed, 91 insertions(+), 56 deletions(-) rename sqlx-pool-guard/src/{macos.rs => apple.rs} (97%) diff --git a/Cargo.lock b/Cargo.lock index 67d64a8402..41ac3ca38f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", ] diff --git a/Cargo.toml b/Cargo.toml index d0033f236a..df80a0c3e6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/common/client-core/surb-storage/Cargo.toml b/common/client-core/surb-storage/Cargo.toml index 49dfbe4b8c..449ce0d589 100644 --- a/common/client-core/surb-storage/Cargo.toml +++ b/common/client-core/surb-storage/Cargo.toml @@ -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] diff --git a/common/client-core/surb-storage/src/backend/fs_backend/manager.rs b/common/client-core/surb-storage/src/backend/fs_backend/manager.rs index 3daa4744a4..68d21009c5 100644 --- a/common/client-core/surb-storage/src/backend/fs_backend/manager.rs +++ b/common/client-core/surb-storage/src/backend/fs_backend/manager.rs @@ -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 { 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 { 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 { 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 { 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, 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, 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, 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(()) } } diff --git a/common/credential-storage/Cargo.toml b/common/credential-storage/Cargo.toml index b822d49db6..07a01486df 100644 --- a/common/credential-storage/Cargo.toml +++ b/common/credential-storage/Cargo.toml @@ -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"] \ No newline at end of file +persistent-storage = ["bincode", "serde"] diff --git a/common/credential-storage/src/backends/sqlite.rs b/common/credential-storage/src/backends/sqlite.rs index 3d91741ee3..6ac45aec47 100644 --- a/common/credential-storage/src/backends/sqlite.rs +++ b/common/credential-storage/src/backends/sqlite.rs @@ -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, 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(()) } diff --git a/common/credential-storage/src/persistent_storage/mod.rs b/common/credential-storage/src/persistent_storage/mod.rs index bccbccf4ab..c8dd5a08b2 100644 --- a/common/credential-storage/src/persistent_storage/mod.rs +++ b/common/credential-storage/src/persistent_storage/mod.rs @@ -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), }) } } diff --git a/sqlx-pool-guard/src/macos.rs b/sqlx-pool-guard/src/apple.rs similarity index 97% rename from sqlx-pool-guard/src/macos.rs rename to sqlx-pool-guard/src/apple.rs index 1142ffd11b..549839a550 100644 --- a/sqlx-pool-guard/src/macos.rs +++ b/sqlx-pool-guard/src/apple.rs @@ -1,6 +1,8 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use std::{io, path::Path}; + use proc_pidinfo::{ ProcFDInfo, ProcFDType, VnodeFdInfoWithPath, proc_pidfdinfo_self, proc_pidinfo_list_self, }; diff --git a/sqlx-pool-guard/src/lib.rs b/sqlx-pool-guard/src/lib.rs index 62838ac20f..b851f1ef79 100644 --- a/sqlx-pool-guard/src/lib.rs +++ b/sqlx-pool-guard/src/lib.rs @@ -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) diff --git a/sqlx-pool-guard/src/linux.rs b/sqlx-pool-guard/src/linux.rs index 5a1fc94bd4..f7610e67c5 100644 --- a/sqlx-pool-guard/src/linux.rs +++ b/sqlx-pool-guard/src/linux.rs @@ -1,6 +1,8 @@ // Copyright 2025 - Nym Technologies SA // 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.