17807091342

Committed 17 Sep 2025 06:28PM UTC coverage: 60.975% (+1.1%) from 59.9%

Build # 17807091342

Build Type

push

github

Committed by

web-flow

Commit Message

fix: always cancel transactions (#7500)

Description
---
Cancel tx and outputs. Don't assume they exist or not. Sending 1-sided
offline transactions doesn't have an existing transaction to cancel

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Improved reliability of canceling pending transactions: missing or
already-processed pending records and unlock failures are now logged as
warnings and do not abort cancellation.
* Reduces spurious errors during cancellation while preserving normal
cleanup, signals, and event emission for more stable wallet behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Run Details

0 of 12 new or added lines in 1 file covered. (0.0%)

493 existing lines in 17 files now uncovered.

74571 of 122298 relevant lines covered (60.97%)

223264.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.63

/common_sqlite/src/sqlite_connection_pool.rs

// Copyright 2020. The Tari Project
//
// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
// following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
// disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
// following disclaimer in the documentation and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
// products derived from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

use core::time::Duration;
use std::{convert::TryFrom, fmt::Display, path::PathBuf};

use diesel::{
    r2d2::{ConnectionManager, Pool, PooledConnection},
    SqliteConnection,
};
use log::*;

use crate::{connection_options::ConnectionOptions, error::SqliteStorageError};

const LOG_TARGET: &str = "common_sqlite::sqlite_connection_pool";

/// The default timeout for acquiring an R2D2 pool connection over and above the PRAGMA busy timeout.
pub const R2D2_POOL_CONNECTION_DELTA: Duration = Duration::from_secs(5);
/// The default timeout for acquiring an R2D2 pool connection.
/// Note: The default R2D2 connection timeout is 30s.
pub const R2D2_POOL_CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);

/// Thin wrapper around an r2d2 `Pool<SqliteConnection>` with standard SQLite
/// configuration (WAL, FKs, busy timeout) applied to each connection.
///
/// The pool is created lazily by calling [`create_pool`] once. After that,
/// callers can obtain connections via [`get_pooled_connection`], a timed
/// variant, or a non-blocking `try_get_pooled_connection`.
///
/// # Concurrency notes
/// SQLite allows only a single writer at a time (even in WAL mode). Keep write
/// transactions short and consider limiting concurrent writers at the
/// application layer (e.g. a semaphore) to reduce lock contention.
///
/// # Timeout interplay
/// The r2d2 **pool checkout** timeout is configured via
/// `Pool::builder().connection_timeout(...)` inside [`create_pool`]. This is
/// separate from SQLite’s `PRAGMA busy_timeout` set by [`ConnectionOptions`].
/// Prefer `connection_timeout >= busy_timeout` (plus a little headroom) to
/// avoid premature pool timeouts while connections are waiting on SQLite locks.
#[derive(Clone)]
pub struct SqliteConnectionPool {
    /// The underlying r2d2 pool. `None` until [`create_pool`] is called.
    pool: Option<Pool<ConnectionManager<SqliteConnection>>>,
    /// Database path / connection string (`:memory:`, filesystem path, or
    /// `file:NAME?mode=memory&cache=shared`).
    db_path: String,
    /// Maximum number of concurrently open connections managed by the pool.
    pool_size: usize,
    /// Per-connection SQLite options applied on acquisition.
    connection_options: ConnectionOptions,
}

impl SqliteConnectionPool {
    /// Create a wrapper with the given target database, pool size and
    /// connection options (WAL, FKs, busy timeout). The r2d2 pool is not built
    /// until [`create_pool`] is called.
    pub fn new(
        db_path: String,
        pool_size: usize,
        enable_wal: bool,
        enable_foreign_keys: bool,
        busy_timeout: Duration,
    ) -> Self {
        Self {
            pool: None,
            db_path,
            pool_size,
            connection_options: ConnectionOptions::new(enable_wal, enable_foreign_keys, busy_timeout),
        }
    }

    /// Create an sqlite connection pool managed by the pool connection manager
    pub fn create_pool(&mut self) -> Result<(), SqliteStorageError> {
        if self.pool.is_none() {
            let mut builder = Pool::builder()
                .max_size(u32::try_from(self.pool_size)?)
                .connection_customizer(Box::new(self.connection_options.clone()));
            if let Some(timeout) = self.connection_options.get_busy_timeout() {
                // When we get a pooled connection, we want the pool connection timeout to be longer
                // than the database busy timeout (set by `PRAGMA busy_timeout`). Here we set the
                // connection timeout to whatever the busy timeout is plus a delta (5s).
                builder = builder.connection_timeout(timeout + R2D2_POOL_CONNECTION_DELTA);
            } else {
                // If no busy timeout is set, we use the default value.
                builder = builder.connection_timeout(R2D2_POOL_CONNECTION_TIMEOUT);
            }
            let pool = builder
                .build(ConnectionManager::<SqliteConnection>::new(self.db_path.as_str()))
                .map_err(|e| SqliteStorageError::DieselR2d2Error(e.to_string()))?;
            self.pool = Some(pool);
        } else {
            warn!(target: LOG_TARGET, "Connection pool for {} already exists", self.db_path);
        }
        Ok(())
    }

    /// Return a pooled sqlite connection managed by the pool connection manager, waits for at most the configured
    /// connection timeout before returning an error.
    pub fn get_pooled_connection(
        &self,
    ) -> Result<PooledConnection<ConnectionManager<SqliteConnection>>, SqliteStorageError> {
        if let Some(pool) = self.pool.as_ref() {
            let start = std::time::Instant::now();
            let connection = pool.get().map_err(|e| {
                warn!(target: LOG_TARGET, "Connection pool state {:?}: {}", pool.state(), e);
                SqliteStorageError::DieselR2d2Error(e.to_string())
            });
            trace!(target: LOG_TARGET, "Acquired 'get_pooled_connection' from pool in {:.2?}", start.elapsed());
            connection
        } else {
            Err(SqliteStorageError::DieselR2d2Error("Pool does not exist".to_string()))
        }
    }

    /// Return a pooled sqlite connection managed by the pool connection manager, waits for at most supplied
    /// connection timeout before returning an error.
    pub fn get_pooled_connection_timeout(
        &self,
        timeout: Duration,
    ) -> Result<PooledConnection<ConnectionManager<SqliteConnection>>, SqliteStorageError> {
        if let Some(pool) = self.pool.clone() {
            let start = std::time::Instant::now();
            let connection = pool.get_timeout(timeout).map_err(|e| {
                warn!(target: LOG_TARGET, "Connection pool state {:?}: {}", pool.state(), e);
                SqliteStorageError::DieselR2d2Error(e.to_string())
            });
            trace!(target: LOG_TARGET, "Acquired 'get_pooled_connection_timeout' from pool in {:.2?}", start.elapsed());
            connection
        } else {
            Err(SqliteStorageError::DieselR2d2Error("Pool does not exist".to_string()))
        }
    }

    /// Return a pooled sqlite connection managed by the pool connection manager, returns None if there are no idle
    /// connections available in the pool. This method will not block waiting to establish a new connection.
    pub fn try_get_pooled_connection(
        &self,
    ) -> Result<Option<PooledConnection<ConnectionManager<SqliteConnection>>>, SqliteStorageError> {
        if let Some(pool) = self.pool.clone() {
            let start = std::time::Instant::now();
            let connection = pool.try_get();
            if connection.is_none() {
                warn!(target: LOG_TARGET, "No connections available, pool state {:?}", pool.state());
            } else {
                trace!(target: LOG_TARGET, "Acquired 'try_get_pooled_connection' from pool in {:.2?}", start.elapsed());
            }
            Ok(connection)
        } else {
            Err(SqliteStorageError::DieselR2d2Error("Pool does not exist".to_string()))
        }
    }

    /// Returns the database path / connection string used by this pool.
    pub fn db_path(&self) -> PathBuf {
        PathBuf::from(&self.db_path)
    }

    /// Perform cleanup on the connection pool. This will drop the pool and return the state of the pool.
    pub fn cleanup(&mut self) -> Option<String> {
        if let Some(pool) = self.pool.take() {
            let state = format!("{:?}", pool.state());
            drop(pool);
            return Some(state);
        }
        None
    }
}

impl Display for SqliteConnectionPool {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let pool_state = if let Some(pool) = self.pool.clone() {
            format!("{:?}", pool.state())
        } else {
            "None".to_string()
        };
        write!(
            f,
            "SqliteConnectionPool {{ pool state: {}, db_path: {}, pool_size: {}, connection_options: {:?} }}",
            pool_state, self.db_path, self.pool_size, self.connection_options
        )
    }
}

/// Helper trait for components that need a pooled SQLite connection.
pub trait PooledDbConnection: Send + Sync + Clone {
    /// Acquire a pooled connection, or return an error if the pool is
    /// unavailable or the checkout times out.
    type Error;

    fn get_pooled_connection(&self) -> Result<PooledConnection<ConnectionManager<SqliteConnection>>, Self::Error>;
}

1	// Copyright 2020. The Tari Project
2	//
3	// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
4	// following conditions are met:
5	//
6	// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
7	// disclaimer.
8	//
9	// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
10	// following disclaimer in the documentation and/or other materials provided with the distribution.
11	//
12	// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
13	// products derived from this software without specific prior written permission.
14	//
15	// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
16	// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
17	// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
18	// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
19	// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
20	// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
21	// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
22
23	use core::time::Duration;
24	use std::{convert::TryFrom, fmt::Display, path::PathBuf};
25
26	use diesel::{
27	r2d2::{ConnectionManager, Pool, PooledConnection},
28	SqliteConnection,
29	};
30	use log::*;
31
32	use crate::{connection_options::ConnectionOptions, error::SqliteStorageError};
33
34	const LOG_TARGET: &str = "common_sqlite::sqlite_connection_pool";
35
36	/// The default timeout for acquiring an R2D2 pool connection over and above the PRAGMA busy timeout.
37	pub const R2D2_POOL_CONNECTION_DELTA: Duration = Duration::from_secs(5);
38	/// The default timeout for acquiring an R2D2 pool connection.
39	/// Note: The default R2D2 connection timeout is 30s.
40	pub const R2D2_POOL_CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
41
42	/// Thin wrapper around an r2d2 `Pool<SqliteConnection>` with standard SQLite
43	/// configuration (WAL, FKs, busy timeout) applied to each connection.
44	///
45	/// The pool is created lazily by calling [`create_pool`] once. After that,
46	/// callers can obtain connections via [`get_pooled_connection`], a timed
47	/// variant, or a non-blocking `try_get_pooled_connection`.
48	///
49	/// # Concurrency notes
50	/// SQLite allows only a single writer at a time (even in WAL mode). Keep write
51	/// transactions short and consider limiting concurrent writers at the
52	/// application layer (e.g. a semaphore) to reduce lock contention.
53	///
54	/// # Timeout interplay
55	/// The r2d2 pool checkout timeout is configured via
56	/// `Pool::builder().connection_timeout(...)` inside [`create_pool`]. This is
57	/// separate from SQLite’s `PRAGMA busy_timeout` set by [`ConnectionOptions`].
58	/// Prefer `connection_timeout >= busy_timeout` (plus a little headroom) to
59	/// avoid premature pool timeouts while connections are waiting on SQLite locks.
60	#[derive(Clone)]
61	pub struct SqliteConnectionPool {
62	/// The underlying r2d2 pool. `None` until [`create_pool`] is called.
63	pool: Option<Pool<ConnectionManager<SqliteConnection>>>,
64	/// Database path / connection string (`:memory:`, filesystem path, or
65	/// `file:NAME?mode=memory&cache=shared`).
66	db_path: String,
67	/// Maximum number of concurrently open connections managed by the pool.
68	pool_size: usize,
69	/// Per-connection SQLite options applied on acquisition.
70	connection_options: ConnectionOptions,
71	}
72
73	impl SqliteConnectionPool {
74	/// Create a wrapper with the given target database, pool size and
75	/// connection options (WAL, FKs, busy timeout). The r2d2 pool is not built
76	/// until [`create_pool`] is called.
77	pub fn new(	340✔
78	db_path: String,	340✔
79	pool_size: usize,	340✔
80	enable_wal: bool,	340✔
81	enable_foreign_keys: bool,	340✔
82	busy_timeout: Duration,	340✔
83	) -> Self {	340✔
84	Self {	340✔
85	pool: None,	340✔
86	db_path,	340✔
87	pool_size,	340✔
88	connection_options: ConnectionOptions::new(enable_wal, enable_foreign_keys, busy_timeout),	340✔
89	}	340✔
90	}	340✔
91
92	/// Create an sqlite connection pool managed by the pool connection manager
93	pub fn create_pool(&mut self) -> Result<(), SqliteStorageError> {	340✔
94	if self.pool.is_none() {	340✔
95	let mut builder = Pool::builder()	340✔
96	.max_size(u32::try_from(self.pool_size)?)	340✔
97	.connection_customizer(Box::new(self.connection_options.clone()));	340✔
98	if let Some(timeout) = self.connection_options.get_busy_timeout() {	340✔
99	// When we get a pooled connection, we want the pool connection timeout to be longer	340✔
100	// than the database busy timeout (set by `PRAGMA busy_timeout`). Here we set the	340✔
101	// connection timeout to whatever the busy timeout is plus a delta (5s).	340✔
102	builder = builder.connection_timeout(timeout + R2D2_POOL_CONNECTION_DELTA);	340✔
103	} else {	340✔
104	// If no busy timeout is set, we use the default value.	×
105	builder = builder.connection_timeout(R2D2_POOL_CONNECTION_TIMEOUT);	×
106	}	×
107	let pool = builder	340✔
108	.build(ConnectionManager::<SqliteConnection>::new(self.db_path.as_str()))	340✔
109	.map_err(\|e\| SqliteStorageError::DieselR2d2Error(e.to_string()))?;	340✔
110	self.pool = Some(pool);	340✔
111	} else {
112	warn!(target: LOG_TARGET, "Connection pool for {} already exists", self.db_path);	×
113	}
114	Ok(())	340✔
115	}	340✔
116
117	/// Return a pooled sqlite connection managed by the pool connection manager, waits for at most the configured
118	/// connection timeout before returning an error.
119	pub fn get_pooled_connection(	17,795✔
120	&self,	17,795✔
121	) -> Result<PooledConnection<ConnectionManager<SqliteConnection>>, SqliteStorageError> {	17,795✔
122	if let Some(pool) = self.pool.as_ref() {	17,795✔
123	let start = std::time::Instant::now();	17,795✔
124	let connection = pool.get().map_err(\|e\| {	17,795✔
125	warn!(target: LOG_TARGET, "Connection pool state {:?}: {}", pool.state(), e);	×
126	SqliteStorageError::DieselR2d2Error(e.to_string())	×
127	});	17,795✔
128	trace!(target: LOG_TARGET, "Acquired 'get_pooled_connection' from pool in {:.2?}", start.elapsed());	17,795✔
129	connection	17,795✔
130	} else {
UNCOV 131	Err(SqliteStorageError::DieselR2d2Error("Pool does not exist".to_string()))	×
132	}
133	}	17,795✔
134
135	/// Return a pooled sqlite connection managed by the pool connection manager, waits for at most supplied
136	/// connection timeout before returning an error.
137	pub fn get_pooled_connection_timeout(	×
UNCOV 138	&self,	×
UNCOV 139	timeout: Duration,	×
UNCOV 140	) -> Result<PooledConnection<ConnectionManager<SqliteConnection>>, SqliteStorageError> {	×
UNCOV 141	if let Some(pool) = self.pool.clone() {	×
UNCOV 142	let start = std::time::Instant::now();	×
UNCOV 143	let connection = pool.get_timeout(timeout).map_err(\|e\| {	×
UNCOV 144	warn!(target: LOG_TARGET, "Connection pool state {:?}: {}", pool.state(), e);	×
UNCOV 145	SqliteStorageError::DieselR2d2Error(e.to_string())	×
UNCOV 146	});	×
UNCOV 147	trace!(target: LOG_TARGET, "Acquired 'get_pooled_connection_timeout' from pool in {:.2?}", start.elapsed());	×
UNCOV 148	connection	×
149	} else {
150	Err(SqliteStorageError::DieselR2d2Error("Pool does not exist".to_string()))	×
151	}
UNCOV 152	}	×
153
154	/// Return a pooled sqlite connection managed by the pool connection manager, returns None if there are no idle
155	/// connections available in the pool. This method will not block waiting to establish a new connection.
UNCOV 156	pub fn try_get_pooled_connection(	×
UNCOV 157	&self,	×
UNCOV 158	) -> Result<Option<PooledConnection<ConnectionManager<SqliteConnection>>>, SqliteStorageError> {	×
UNCOV 159	if let Some(pool) = self.pool.clone() {	×
UNCOV 160	let start = std::time::Instant::now();	×
UNCOV 161	let connection = pool.try_get();	×
UNCOV 162	if connection.is_none() {	×
UNCOV 163	warn!(target: LOG_TARGET, "No connections available, pool state {:?}", pool.state());	×
164	} else {
UNCOV 165	trace!(target: LOG_TARGET, "Acquired 'try_get_pooled_connection' from pool in {:.2?}", start.elapsed());	×
166	}
UNCOV 167	Ok(connection)	×
168	} else {
UNCOV 169	Err(SqliteStorageError::DieselR2d2Error("Pool does not exist".to_string()))	×
170	}
UNCOV 171	}	×
172
173	/// Returns the database path / connection string used by this pool.
174	pub fn db_path(&self) -> PathBuf {	2,529✔
175	PathBuf::from(&self.db_path)	2,529✔
176	}	2,529✔
177
178	/// Perform cleanup on the connection pool. This will drop the pool and return the state of the pool.
179	pub fn cleanup(&mut self) -> Option<String> {	189✔
180	if let Some(pool) = self.pool.take() {	189✔
181	let state = format!("{:?}", pool.state());	189✔
182	drop(pool);	189✔
183	return Some(state);	189✔
UNCOV 184	}	×
UNCOV 185	None	×
186	}	189✔
187	}
188
189	impl Display for SqliteConnectionPool {
UNCOV 190	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	×
UNCOV 191	let pool_state = if let Some(pool) = self.pool.clone() {	×
UNCOV 192	format!("{:?}", pool.state())	×
193	} else {
UNCOV 194	"None".to_string()	×
195	};
UNCOV 196	write!(	×
UNCOV 197	f,	×
UNCOV 198	"SqliteConnectionPool {{ pool state: {}, db_path: {}, pool_size: {}, connection_options: {:?} }}",	×
UNCOV 199	pool_state, self.db_path, self.pool_size, self.connection_options	×
UNCOV 200	)	×
UNCOV 201	}	×
202	}
203
204	/// Helper trait for components that need a pooled SQLite connection.
205	pub trait PooledDbConnection: Send + Sync + Clone {
206	/// Acquire a pooled connection, or return an error if the pool is
207	/// unavailable or the checkout times out.
208	type Error;
209
210	fn get_pooled_connection(&self) -> Result<PooledConnection<ConnectionManager<SqliteConnection>>, Self::Error>;
211	}

tari-project / tari / 17807091342

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous