diff --git a/Cargo.lock b/Cargo.lock index 4cad468c66..13a80c207b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7954,6 +7954,15 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proc_pidinfo" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1724fdb3f7349e4c58806dfb29dbc3b5e2f8b0997bb5117acf8db6e555867a91" +dependencies = [ + "libc", +] + [[package]] name = "prometheus" version = "0.14.0" @@ -9525,6 +9534,16 @@ dependencies = [ "whoami", ] +[[package]] +name = "sqlx-pool-guard" +version = "0.1.0" +dependencies = [ + "log", + "proc_pidinfo", + "sqlx", + "tokio", +] + [[package]] name = "sqlx-postgres" version = "0.8.6" diff --git a/Cargo.toml b/Cargo.toml index c2f5b97078..e15ac95c90 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -124,7 +124,7 @@ members = [ "service-providers/authenticator", "service-providers/common", "service-providers/ip-packet-router", - "service-providers/network-requester", + "service-providers/network-requester", "sqlx-pool-guard", "tools/echo-server", "tools/internal/contract-state-importer/importer-cli", "tools/internal/contract-state-importer/importer-contract", @@ -286,6 +286,7 @@ petgraph = "0.6.5" pin-project = "1.1" pin-project-lite = "0.2.16" publicsuffix = "2.3.0" +proc_pidinfo = "0.1" quote = "1" rand = "0.8.5" rand_chacha = "0.3" diff --git a/sqlx-pool-guard/Cargo.toml b/sqlx-pool-guard/Cargo.toml new file mode 100644 index 0000000000..ee60690b4d --- /dev/null +++ b/sqlx-pool-guard/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "sqlx-pool-guard" +version = "0.1.0" +edition = "2024" +license.workspace = true + +[lints] +workspace = true + +[dependencies] +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "time", "fs"] } +sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] } +log.workspace = true + +[target.'cfg(target_os = "macos")'.dependencies] +proc_pidinfo.workspace = true diff --git a/sqlx-pool-guard/src/lib.rs b/sqlx-pool-guard/src/lib.rs new file mode 100644 index 0000000000..05a933fe5f --- /dev/null +++ b/sqlx-pool-guard/src/lib.rs @@ -0,0 +1,201 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use std::{ + io, + ops::{Deref, DerefMut}, + path::{Path, PathBuf}, + time::Duration, +}; + +#[cfg(target_os = "macos")] +use proc_pidinfo::{ + ProcFDInfo, ProcFDType, VnodeFdInfoWithPath, proc_pidfdinfo_self, proc_pidinfo_list_self, +}; + +const SQL_CLOSE_MAX_ATTEMPTS: u8 = 10; +const SQL_CLOSE_RETRY_DELAY: Duration = Duration::from_millis(100); + +pub struct SqlitePoolGuard { + database_path: PathBuf, + connection_pool: sqlx::SqlitePool, +} + +impl Deref for SqlitePoolGuard { + type Target = sqlx::SqlitePool; + fn deref(&self) -> &Self::Target { + &self.connection_pool + } +} + +impl DerefMut for SqlitePoolGuard { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.connection_pool + } +} + +impl SqlitePoolGuard { + pub fn new(database_path: PathBuf, connection_pool: sqlx::SqlitePool) -> Self { + Self { + database_path, + connection_pool, + } + } + + /// Close udnerlying sqlite pool + pub async fn close_pool(&self) { + _ = self.close_pool_inner(); + } + + async fn close_pool_inner(&self) -> std::io::Result<()> { + self.connection_pool.close().await; + + #[cfg(target_os = "macos")] + if let Err(e) = self.wait_io_close().await { + log::error!("Failed to wait for file to close: {e}"); + } + + Ok(()) + } + + /// Returns all database files, including shm and wal files. + fn all_database_files(&self) -> Vec { + let mut database_files = vec![]; + let canonical_path = self + .database_path + .canonicalize() + .inspect_err(|e| { + log::error!( + "Failed to canonicalize path: {}. Cause: {e}", + self.database_path.display() + ); + }) + .unwrap_or(self.database_path.clone()); + + if let Some(ext) = canonical_path.extension() { + for added_ext in ["-shm", "-wal"] { + let mut new_ext = ext.to_owned(); + new_ext.push(added_ext); + database_files.push(canonical_path.with_extension(new_ext)); + } + } + database_files.push(canonical_path); + database_files + } + + /// Wait for I/O close to the database files + async fn wait_io_close(&self) -> std::io::Result<()> { + let database_files = self.all_database_files(); + let paths: Vec<&Path> = database_files.iter().map(PathBuf::as_path).collect(); + + for _ in 0..SQL_CLOSE_MAX_ATTEMPTS { + match Self::check_io_close(&paths) + .await + .inspect_err(|e| log::error!("check_io_close() failure: {e}")) + { + Ok(true) | Err(_) => tokio::time::sleep(SQL_CLOSE_RETRY_DELAY).await, + Ok(false) => return Ok(()), + } + } + + Err(io::Error::new( + io::ErrorKind::TimedOut, + "timed out waiting for sqlite files to be closed", + )) + } + + /// Check if no more open file descriptors exist for the given files. + /// + /// On macOS this is done using `proc_pidinfo` (`sys/proc_info.h`) + /// See: http://blog.palominolabs.com/2012/06/19/getting-the-files-being-used-by-a-process-on-mac-os-x/ + #[cfg(target_os = "macos")] + async fn check_io_close(file_paths: &[&Path]) -> io::Result { + let fd_list = proc_pidinfo_list_self::()?; + + for fd in fd_list + .iter() + .filter(|s| s.fd_type() == Ok(ProcFDType::VNODE)) + { + let Some(vnode) = proc_pidfdinfo_self::(fd.proc_fd) + .inspect_err(|e| { + log::warn!("proc_pidfdinfo_self::() failure: {e}"); + }) + .ok() + .flatten() + else { + continue; + }; + + if let Ok(true) = vnode + .path() + .map(|vnode_path| file_paths.contains(&vnode_path)) + .inspect_err(|e| { + log::warn!("vnode.path() failure: {e:?}"); + }) + { + return Ok(true); + } + } + + Ok(false) + } + + /// Check if no more open file descriptors exist for the given files. + #[cfg(target_os = "linux")] + async fn check_io_close(file_paths: &[&Path]) -> io::Result { + let mut dir = tokio::fs::read_dir("/proc/self/fd/").await?; + + while let Ok(Some(entry)) = dir.next_entry().await { + // Looking for DT_LNK + if entry + .file_type() + .await + .inspect_err(|e| log::warn!("entry.file_type() failure: {e}")) + .is_ok_and(|entry_type| entry_type.is_symlink()) + { + // atoi(d->d_name) to obtain open fd? + let path = PathBuf::from(entry.file_name()); + if file_paths.contains(&path.as_ref()) { + return Ok(true); + } + } + } + + Ok(false) + } +} + +#[cfg(test)] +mod tests { + use sqlx::{ + ConnectOptions, + sqlite::{SqliteAutoVacuum, SqliteSynchronous}, + }; + + use super::*; + + #[tokio::test] + async fn test_wait_close() { + let database_path = PathBuf::from("/tmp/storage.sqlite"); + let opts = sqlx::sqlite::SqliteConnectOptions::new() + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .auto_vacuum(SqliteAutoVacuum::Incremental) + .filename(database_path.clone()) + .create_if_missing(true) + .disable_statement_logging(); + let connection_pool = sqlx::SqlitePool::connect_with(opts).await.unwrap(); + + let guard = SqlitePoolGuard::new(database_path, connection_pool); + + assert!( + guard + .wait_io_close() + .await + .err() + .is_some_and(|e| e.kind() == io::ErrorKind::TimedOut) + ); + + assert!(guard.close_pool_inner().await.is_ok()); + } +}