miden_node_db/
lib.rs

1mod conv;
2mod errors;
3mod manager;
4pub mod migration;
5pub mod sqlite;
6
7use std::num::NonZeroUsize;
8use std::path::Path;
9
10pub use conv::{DatabaseTypeConversionError, SqlTypeConvert};
11use diesel::{RunQueryDsl, SqliteConnection};
12pub use errors::{DatabaseError, SchemaVerificationError};
13pub use manager::{ConnectionManager, ConnectionManagerError, configure_connection_on_creation};
14use tracing::Instrument;
15
16pub type Result<T, E = DatabaseError> = std::result::Result<T, E>;
17
18/// Returns the default SQLite connection pool size.
19///
20/// Defaults to twice the available CPU parallelism. If the OS cannot report the available
21/// parallelism, fall back to two connections.
22pub fn default_connection_pool_size() -> NonZeroUsize {
23    let available_cores = std::thread::available_parallelism().map_or(1, NonZeroUsize::get);
24    let connection_count = available_cores.saturating_mul(2);
25    NonZeroUsize::new(connection_count).expect("connection count must be non-zero")
26}
27
28/// Database handle that provides fundamental operations that various components of Miden Node can
29/// utililze for their storage needs.
30#[derive(Clone)]
31pub struct Db {
32    pool: deadpool_diesel::Pool<ConnectionManager, deadpool::managed::Object<ConnectionManager>>,
33}
34
35impl Db {
36    /// Creates a new database instance with the provided connection pool.
37    pub fn new(database_filepath: &Path) -> Result<Self, DatabaseError> {
38        Self::new_with_pool_size(database_filepath, default_connection_pool_size())
39    }
40
41    /// Creates a new database instance with the provided connection pool size.
42    pub fn new_with_pool_size(
43        database_filepath: &Path,
44        connection_pool_size: NonZeroUsize,
45    ) -> Result<Self, DatabaseError> {
46        let manager = ConnectionManager::new(database_filepath.to_str().unwrap());
47        let pool = deadpool_diesel::Pool::builder(manager)
48            .max_size(connection_pool_size.get())
49            .build()?;
50        Ok(Self { pool })
51    }
52
53    /// Checks out a connection from the pool and pins it for the caller's exclusive, long-lived
54    /// use. See [`PinnedConnection`].
55    ///
56    /// This removes one connection from the shared pool for the lifetime of the returned handle,
57    /// so the pool must be sized to leave at least one connection for other users.
58    pub async fn pinned_connection(&self) -> Result<PinnedConnection, DatabaseError> {
59        let conn = self
60            .pool
61            .get()
62            .in_current_span()
63            .await
64            .map_err(|e| DatabaseError::ConnectionPoolObtainError(Box::new(e)))?;
65        Ok(PinnedConnection { conn })
66    }
67
68    /// Create and commit a transaction with the queries added in the provided closure
69    pub async fn transact<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
70    where
71        Q: Send
72            + for<'a, 't> FnOnce(&'a mut SqliteConnection) -> std::result::Result<R, E>
73            + 'static,
74        R: Send + 'static,
75        M: Send + ToString,
76        E: From<diesel::result::Error>,
77        E: From<DatabaseError>,
78        E: std::error::Error + Send + Sync + 'static,
79    {
80        self.pinned_connection().await.map_err(E::from)?.transact(msg, query).await
81    }
82
83    /// Run the query _without_ a transaction
84    pub async fn query<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
85    where
86        Q: Send + FnOnce(&mut SqliteConnection) -> std::result::Result<R, E> + 'static,
87        R: Send + 'static,
88        M: Send + ToString,
89        E: From<DatabaseError>,
90        E: std::error::Error + Send + Sync + 'static,
91    {
92        self.pinned_connection().await.map_err(E::from)?.query(msg, query).await
93    }
94}
95
96/// A connection checked out of [`Db`]'s pool and held for the caller's exclusive, long-lived use.
97///
98/// A hot event loop can pin a connection so its queries never wait on the shared pool even when
99/// many concurrent tasks are saturating it. `transact`/`query` mirror [`Db`]'s, but run on the
100/// pinned connection rather than acquiring one per call. The connection is returned to the pool
101/// when the `PinnedConnection` is dropped.
102pub struct PinnedConnection {
103    conn: deadpool::managed::Object<ConnectionManager>,
104}
105
106impl PinnedConnection {
107    /// Create and commit a transaction with the queries added in the provided closure, running on
108    /// the pinned connection.
109    pub async fn transact<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
110    where
111        Q: Send
112            + for<'a, 't> FnOnce(&'a mut SqliteConnection) -> std::result::Result<R, E>
113            + 'static,
114        R: Send + 'static,
115        M: Send + ToString,
116        E: From<diesel::result::Error>,
117        E: From<DatabaseError>,
118        E: std::error::Error + Send + Sync + 'static,
119    {
120        let span = tracing::Span::current();
121        self.conn
122            .interact(move |conn| {
123                let _guard = span.enter();
124                <_ as diesel::Connection>::transaction::<R, E, Q>(conn, query)
125            })
126            .await
127            .map_err(|err| E::from(DatabaseError::interact(&msg.to_string(), &err)))?
128    }
129
130    /// Run the query _without_ a transaction on the pinned connection.
131    pub async fn query<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
132    where
133        Q: Send + FnOnce(&mut SqliteConnection) -> std::result::Result<R, E> + 'static,
134        R: Send + 'static,
135        M: Send + ToString,
136        E: From<DatabaseError>,
137        E: std::error::Error + Send + Sync + 'static,
138    {
139        let span = tracing::Span::current();
140        self.conn
141            .interact(move |conn| {
142                let _guard = span.enter();
143                query(conn)
144            })
145            .await
146            .map_err(|err| E::from(DatabaseError::interact(&msg.to_string(), &err)))?
147    }
148}
miden_node_db/lib.rs

miden_node_db/
lib.rs
