p2panda_store/sqlite.rs
1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! SQLite database implementation with associated utility functions.
4use std::sync::Arc;
5use std::time::Duration;
6
7use p2panda_core::cbor::EncodeError;
8use sqlx::migrate::{MigrateDatabase, Migrator};
9use sqlx::sqlite::SqlitePoolOptions;
10use sqlx::{Sqlite, migrate};
11use thiserror::Error;
12use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore};
13
14/// Creates the SQLite database if it doesn't already exist.
15pub async fn create_database(url: &str) -> Result<(), SqliteError> {
16 if !Sqlite::database_exists(url).await? {
17 Sqlite::create_database(url).await?
18 }
19 Ok(())
20}
21
22/// Drops the SQLite database if it exists.
23pub async fn drop_database(url: &str) -> Result<(), SqliteError> {
24 if Sqlite::database_exists(url).await? {
25 Sqlite::drop_database(url).await?
26 }
27 Ok(())
28}
29
30/// Creates the SQLite connection pool.
31pub async fn connection_pool(
32 url: &str,
33 max_connections: u32,
34) -> Result<sqlx::SqlitePool, SqliteError> {
35 let pool: sqlx::SqlitePool = SqlitePoolOptions::new()
36 .max_connections(max_connections)
37 .connect(url)
38 .await?;
39 Ok(pool)
40}
41
42/// Gets migrations from folder without running them.
43pub fn migrations() -> Migrator {
44 migrate!()
45}
46
47/// Runs any pending database migrations from inside the application.
48pub async fn run_pending_migrations(pool: &sqlx::SqlitePool) -> Result<(), SqliteError> {
49 migrations().run(pool).await?;
50 Ok(())
51}
52
53/// Builder for `SqliteStore`.
54///
55/// To create the database call `SqliteStoreBuilder::build()`.
56///
57/// By default, the builder configures an in-memory database with a maximum number of 16
58/// connections. The database is created if it doesn't already exist and migrations are
59/// automatically run on start-up.
60pub struct SqliteStoreBuilder {
61 url: String,
62 min_connections: u32,
63 max_connections: u32,
64 idle_timeout: Option<Duration>,
65 max_lifetime: Option<Duration>,
66 run_migrations: bool,
67 create_database: bool,
68}
69
70impl Default for SqliteStoreBuilder {
71 fn default() -> Self {
72 Self {
73 url: ":memory:".into(),
74 min_connections: 3,
75 max_connections: 16,
76 idle_timeout: Some(Duration::from_secs(10 * 60)),
77 max_lifetime: Some(Duration::from_secs(30 * 60)),
78 create_database: true,
79 run_migrations: true,
80 }
81 }
82}
83
84impl SqliteStoreBuilder {
85 /// Creates a new `SqliteStoreBuilder` using default configuration values.
86 pub fn new() -> Self {
87 Self::default()
88 }
89
90 /// Creates a new in-memory `SqliteStoreBuilder` using recommended configuration values.
91 ///
92 /// The configuration values have been chosen to prevent the in-memory database being dropped
93 /// when there are no active connections and the idle timeout or max lifetime limit is reached.
94 pub fn memory() -> Self {
95 Self::default()
96 .database_url(":memory:")
97 .min_connections(1)
98 .max_connections(1)
99 .idle_timeout(None)
100 .max_lifetime(None)
101 }
102
103 /// Sets the database URL.
104 ///
105 /// If left unset, the database will use an ephemeral in-memory URL.
106 pub fn database_url(mut self, url: &str) -> Self {
107 self.url = url.to_string();
108 self
109 }
110
111 /// Sets the minimum number of connections to be maintained by the database pool.
112 ///
113 /// If left unset, a minimum of 3 connections will be maintained.
114 pub fn min_connections(mut self, min_connections: u32) -> Self {
115 self.min_connections = min_connections;
116 self
117 }
118
119 /// Sets the maximum number of connections to be maintained by the database pool.
120 ///
121 /// If left unset, a maximum of 16 connections will be maintained.
122 pub fn max_connections(mut self, max_connections: u32) -> Self {
123 self.max_connections = max_connections;
124 self
125 }
126
127 /// Set a maximum idle duration for individual connections.
128 ///
129 /// Any connection that remains in the idle queue longer than this will be closed.
130 pub fn idle_timeout(mut self, timeout: impl Into<Option<Duration>>) -> Self {
131 self.idle_timeout = timeout.into();
132 self
133 }
134
135 /// Set the maximum lifetime of individual connections.
136 ///
137 /// Any connection with a lifetime greater than this will be closed.
138 ///
139 /// When set to `None`, all connections live until either reaped by [`idle_timeout`] or
140 /// explicitly disconnected.
141 ///
142 /// Infinite connections are not recommended due to the unfortunate reality of memory/resource
143 /// leaks on the database-side. It is better to retire connections periodically (even if only
144 /// once daily) to allow the database the opportunity to clean up data structures (parse trees,
145 /// query metadata caches, thread-local storage, etc.) that are associated with a session.
146 pub fn max_lifetime(mut self, lifetime: impl Into<Option<Duration>>) -> Self {
147 self.max_lifetime = lifetime.into();
148 self
149 }
150
151 /// Creates the database if it doesn't already exist.
152 ///
153 /// If left unset, the database will be created by default.
154 pub fn create_database(mut self, create_database: bool) -> Self {
155 self.create_database = create_database;
156 self
157 }
158
159 /// Sets whether pending migrations should be applied when the database is built.
160 ///
161 /// If left unset, the database will apply any pending migrations.
162 pub fn run_default_migrations(mut self, run_migrations: bool) -> Self {
163 self.run_migrations = run_migrations;
164 self
165 }
166
167 /// Builds the `SqliteStore`.
168 pub async fn build(self) -> Result<SqliteStore, SqliteError> {
169 if self.create_database {
170 create_database(&self.url).await?;
171 }
172
173 let pool: sqlx::SqlitePool = SqlitePoolOptions::new()
174 .min_connections(self.min_connections)
175 .max_connections(self.max_connections)
176 .idle_timeout(self.idle_timeout)
177 .max_lifetime(self.max_lifetime)
178 .connect(&self.url)
179 .await?;
180
181 if self.run_migrations {
182 run_pending_migrations(&pool).await?;
183 }
184
185 Ok(SqliteStore::new(pool))
186 }
187}
188
189/// An in-progress database transaction.
190pub type Transaction<'a> = sqlx::Transaction<'a, Sqlite>;
191
192/// Sqlite connection pool.
193pub type SqlitePool = sqlx::SqlitePool;
194
195/// SQLite database with connection pool and transaction provider.
196///
197/// This struct can be cloned and used in multiple places in the application. Every cloned instance
198/// will re-use the same connection pool and have access to the same transaction instance if one
199/// was started. To guard against sharing transactions unknowingly across unrelated database
200/// queries, a concept of a `TransactionPermit` was introduced which does not protect from misuse
201/// but helps to make "holding" a transaction explicit.
202///
203/// Please note that SQLite strictly serializes transactions with _writes_ and will block any
204/// parallel attempt to begin another one. Processes starting a transaction will acquire a
205/// `TransactionPermit` and keep it until the transaction was committed or rolled back. If the
206/// query only involves _reads_ it is recommended to not use transactions and use the `execute`
207/// method directly as acquiring transactions will potentially block other processes to do work.
208///
209/// ## Design decisions
210///
211/// This storage API design was chosen to make the dynamics of the underlying SQLite database
212/// explicit to avoid potentially introducing subtle bugs. Internally any process can access the
213/// transaction object to do writes and (uncommitted) reads (see "Transaction I" in diagram). Care
214/// is required when designing systems like that as it's still possible to allow concurrent
215/// processes to read and write within the same transaction (for example one process could roll
216/// back the transaction while the other one assumed it will be committed). Usually developers want
217/// to design _writes_ to the database within a transaction if they need consistency and atomicity
218/// guarantees. "Unrelated" queries _can_ be "pooled" in one transaction (for performance reasons
219/// for example) if consistency is guaranteed by all involved processes and the underlying
220/// data-model (see "Transaction II" in diagram).
221///
222/// ```text
223/// Transaction I:
224/// begin ---------------------> commit
225///
226/// Process I:
227/// --> write --> read -->
228///
229/// Transaction II:
230/// begin ----------------------> commit
231///
232/// Process II:
233/// --> write --> write -->
234///
235/// Process III:
236/// --> read --> write --->
237/// ```
238///
239/// Another design decision is to not expose transactions to the high-level storage APIs (similar
240/// to the "Repository Pattern"). Users of the storage methods like `get_operation` (in
241/// `OperationStore`) etc. do _not_ need to explicity deal with transaction objects, as this is
242/// handled internally now. Like this it is possible to separate the "logic" from the "storage"
243/// layer and keep the code clean.
244#[derive(Clone, Debug)]
245pub struct SqliteStore {
246 tx: Arc<Mutex<Option<Transaction<'static>>>>,
247 pub(crate) pool: sqlx::SqlitePool,
248 semaphore: Arc<Semaphore>,
249}
250
251impl SqliteStore {
252 /// Creates a new `SqliteStore` using the provided connection pool.
253 pub(crate) fn new(pool: sqlx::SqlitePool) -> Self {
254 Self {
255 tx: Arc::default(),
256 pool,
257 // SQLite only ever allows _one_ transaction at a time. This might be a repetition of
258 // what sqlx and SQLite do under the hood, but we want to make this behaviour explicit
259 // right from the beginning with this semaphore.
260 semaphore: Arc::new(Semaphore::new(1)),
261 }
262 }
263
264 /// Creates a new `SqliteStore` using the provided connection pool.
265 pub fn from_pool(pool: sqlx::SqlitePool) -> Self {
266 Self::new(pool)
267 }
268
269 /// Returns a reference to the connection pool.
270 pub fn pool(&self) -> &sqlx::SqlitePool {
271 &self.pool
272 }
273
274 /// Builds an in-memory SQLite database for testing purposes.
275 #[cfg(any(test, feature = "test_utils"))]
276 pub async fn temporary() -> Self {
277 SqliteStoreBuilder::memory()
278 .build()
279 .await
280 .expect("migrations succeeded")
281 }
282
283 /// Executes a SQL query within a transaction.
284 ///
285 /// This method will return an error when no transaction is currently given. Make sure to call
286 /// `begin` before.
287 ///
288 /// If the query fails the user probably wants to roll back the transaction and free the
289 /// permit. This is _not_ handled automatically.
290 pub async fn tx<F, R>(&self, f: F) -> Result<R, SqliteError>
291 where
292 F: AsyncFnOnce(&mut Transaction) -> Result<R, SqliteError>,
293 {
294 let mut tx_ref = self.tx.lock().await;
295 let tx = tx_ref.as_mut().ok_or(SqliteError::TransactionMissing)?;
296
297 f(tx).await
298 }
299
300 /// Executes a SQL query directly.
301 pub async fn execute<F, R>(&self, f: F) -> Result<R, SqliteError>
302 where
303 F: AsyncFnOnce(&sqlx::SqlitePool) -> Result<R, SqliteError>,
304 {
305 f(&self.pool).await
306 }
307}
308
309impl crate::traits::Transaction for SqliteStore {
310 type Error = SqliteError;
311
312 type Permit = TransactionPermit;
313
314 /// Begins a transaction.
315 ///
316 /// Transactions are strictly serialized, this is expressed in form of a `TransactionPermit`
317 /// processes need to hold when acquiring access to a new transaction. Any concurrent process
318 /// calling it will await here if there's already another process holding a permit, this will
319 /// potentially "slow down" work and should be carefully used.
320 ///
321 /// Any process with a transaction can now start using the `tx` method to execute writes within
322 /// this transaction or perform uncommitted "dirty" reads on it.
323 ///
324 /// It is usually not necessary to acquire a transaction when the logic only requires committed
325 /// _reads_ to the database. Use `execute` instead.
326 async fn begin(&self) -> Result<TransactionPermit, SqliteError> {
327 // Acquire a permit from the semaphore, it will await if currently another process has the
328 // permit. Here we enforce strict serialization of transactions (similar to what SQLite
329 // does under the hood).
330 let permit = self
331 .semaphore
332 .clone()
333 .acquire_owned()
334 .await
335 .expect("if semaphore is closed then the whole struct is gone as well");
336
337 // Access the transaction object which we've placed behind a Mutex. This lock follows a
338 // different logic and only makes sure that mutable access to it is exclusive _within_ a
339 // process "holding" the transaction permit.
340 let mut tx_ref = self.tx.lock().await;
341 assert!(
342 tx_ref.is_none(),
343 "can't have an already existing transaction after an just-acquired permit"
344 );
345 let tx = self.pool.begin().await?;
346 tx_ref.replace(tx);
347
348 Ok(TransactionPermit::new(permit, self.tx.clone()))
349 }
350
351 /// Rolls back the transaction and with that all uncommitted changes.
352 ///
353 /// This takes the permit and frees it after the rollback has finished. Other processes can now
354 /// begin new transactions.
355 async fn rollback(&self, permit: TransactionPermit) -> Result<(), SqliteError> {
356 let Some(tx) = self.tx.lock().await.take() else {
357 panic!("can't have no transaction without dropping permit first")
358 };
359
360 let result = tx.rollback().await.map_err(SqliteError::Sqlite);
361
362 // Always drop the permit, both on successful rollback and error. This will allow other
363 // processes now to begin a new transaction and acquire the permit.
364 permit.mark_committed_and_drop();
365
366 result
367 }
368
369 /// Commits the transaction.
370 ///
371 /// This takes the permit and frees it after the commit has finished. Other processes can now
372 /// begin new transactions.
373 async fn commit(&self, permit: TransactionPermit) -> Result<(), SqliteError> {
374 let Some(tx) = self.tx.lock().await.take() else {
375 panic!("can't have no transaction without dropping permit first")
376 };
377
378 let result = tx.commit().await.map_err(SqliteError::Sqlite);
379
380 // Always drop the permit, both on successful commit and error. This will allow other
381 // processes now to begin a new transaction and acquire the permit.
382 permit.mark_committed_and_drop();
383
384 result
385 }
386}
387
388/// Locked context marking the lifetime of a single transaction.
389pub struct TransactionPermit {
390 permit: Arc<OwnedSemaphorePermit>,
391 tx: Arc<Mutex<Option<Transaction<'static>>>>,
392 committed: bool,
393}
394
395impl TransactionPermit {
396 /// Creates a new `TransactionPermit` using the given permit and transaction.
397 pub(super) fn new(
398 permit: OwnedSemaphorePermit,
399 tx: Arc<Mutex<Option<Transaction<'static>>>>,
400 ) -> Self {
401 Self {
402 permit: Arc::new(permit),
403 tx,
404 committed: false,
405 }
406 }
407
408 /// Marks the transaction as committed and drops the permit.
409 ///
410 /// In the case that the permit was never used, whether due to an early return or error, the
411 /// transaction is automatically rolled-back to prevent corrupted state.
412 pub(super) fn mark_committed_and_drop(mut self) {
413 self.committed = true;
414 drop(self)
415 }
416}
417
418impl Drop for TransactionPermit {
419 fn drop(&mut self) {
420 // If the permit was never used (due to an early return / error / etc.) we automatically
421 // roll-back the transaction.
422 if !self.committed {
423 let permit = self.permit.clone();
424 let tx = self.tx.clone();
425
426 tokio::spawn(async move {
427 if let Some(tx) = tx.lock().await.take() {
428 let _ = tx.rollback().await;
429 }
430
431 drop(permit); // Semaphore released only after rollback completes.
432 });
433 }
434 }
435}
436
437/// Error when interacting with a SQLite store implementation.
438#[derive(Debug, Error)]
439pub enum SqliteError {
440 /// This is a critical error as it indicates that something is wrong with the usage of this
441 /// API: Queries using transactions can only ever occur if a transaction was started _before_.
442 #[error("tried to interact with inexistant transaction")]
443 TransactionMissing,
444
445 /// SQLite database and connection error.
446 #[error(transparent)]
447 Sqlite(#[from] sqlx::Error),
448
449 /// SQL table schema migration error.
450 #[error(transparent)]
451 Migrate(#[from] sqlx::migrate::MigrateError),
452
453 /// An I/O error occurred while encoding bytes before storing them into the database. This is a
454 /// critical error.
455 #[error("failed encoding '{0}' value before storing to database: {1}")]
456 Encode(String, EncodeError),
457
458 /// Invalid, corrupted data was found in the database. This is a critical error.
459 #[error("could not decode corrupted '{0}' value from database: {1}")]
460 Decode(String, DecodeError),
461}
462
463/// Error decoding value retrieved from a store.
464#[derive(Debug, Error)]
465pub enum DecodeError {
466 #[error(transparent)]
467 DecodeCbor(#[from] p2panda_core::cbor::DecodeError),
468
469 #[error(transparent)]
470 Hash(#[from] p2panda_core::hash::HashError),
471
472 #[error(transparent)]
473 Topic(#[from] p2panda_core::topic::TopicError),
474
475 #[error("parsing from string failed")]
476 FromStr,
477}
478
479#[cfg(test)]
480mod tests {
481 use futures_test::task::noop_context;
482 use sqlx::{Executor, query, query_as, query_scalar};
483 use tokio::pin;
484
485 use crate::sqlite::{SqliteError, SqliteStore};
486 use crate::traits::Transaction;
487
488 #[tokio::test]
489 async fn transaction_provider() {
490 let pool = SqliteStore::temporary().await;
491
492 // Executing with an in-existant transaction should throw error.
493 std::assert_matches!(
494 pool.tx(async |_| Ok(())).await,
495 Err(SqliteError::TransactionMissing)
496 );
497
498 // Starting a new transaction should work.
499 let permit = pool.begin().await.expect("no error");
500
501 // .. attempting to start a second one should make us wait.
502 {
503 let fut = pool.begin();
504 let mut cx = noop_context();
505 pin!(fut);
506 assert!(fut.poll(&mut cx).is_pending());
507 }
508
509 // Using the transaction should work without failure.
510 assert!(pool.tx(async |_| Ok(())).await.is_ok());
511
512 // Committing should work as well.
513 assert!(pool.commit(permit).await.is_ok());
514
515 // .. and now running a transaction should fail.
516 std::assert_matches!(
517 pool.tx(async |_| Ok(())).await,
518 Err(SqliteError::TransactionMissing)
519 );
520 }
521
522 #[tokio::test]
523 async fn early_permit_drop_causing_rollback() {
524 let pool = SqliteStore::temporary().await;
525
526 // Create test-table schema.
527 pool.execute(async |pool| {
528 pool.execute("CREATE TABLE test(x INTEGER)").await?;
529 Ok(())
530 })
531 .await
532 .unwrap();
533
534 let permit = pool.begin().await.unwrap();
535
536 pool.tx(async |tx| {
537 query("INSERT INTO test (x) VALUES (10)")
538 .execute(&mut **tx)
539 .await?;
540 Ok(())
541 })
542 .await
543 .unwrap();
544
545 // Permit was dropped prematurely without committing.
546 drop(permit);
547
548 // It is okay to start another permit.
549 assert!(pool.begin().await.is_ok());
550
551 // The data was not written as the transaction got rolled back.
552 let count: i64 = pool
553 .execute(async |pool| {
554 query_scalar("SELECT COUNT(*) FROM test")
555 .fetch_one(pool)
556 .await
557 .map_err(SqliteError::Sqlite)
558 })
559 .await
560 .unwrap();
561 assert_eq!(count, 0);
562 }
563
564 #[tokio::test]
565 async fn serialized_transactions() {
566 let pool_1 = SqliteStore::temporary().await;
567
568 let pool_2 = pool_1.clone();
569
570 // Create test-table schema.
571 pool_1
572 .execute(async |pool| {
573 pool.execute("CREATE TABLE test(x INTEGER)").await?;
574 Ok(())
575 })
576 .await
577 .unwrap();
578
579 // 1. Pool 1 acquires the permit to run a transaction.
580 let permit_1 = pool_1.begin().await.unwrap();
581
582 // .. parallely Pool 2 also tries to do some work.
583 let handle = tokio::spawn(async move {
584 // Try to acquire a permit, this will "block" for now as pool 1 already is doing
585 // something and we need to wait.
586 let permit_2 = pool_2.begin().await.unwrap();
587
588 // 5. We should see now the previously change made by pool 1.
589 let result = pool_2
590 .tx(async |tx| {
591 let row: (i64,) = query_as("SELECT x FROM test").fetch_one(&mut **tx).await?;
592 Ok(row.0)
593 })
594 .await
595 .unwrap();
596 assert_eq!(result, 5);
597
598 // 6. Change the value to something else.
599 pool_2
600 .tx(async |tx| {
601 query("INSERT INTO test (x) VALUES (10)")
602 .execute(&mut **tx)
603 .await?;
604 Ok(())
605 })
606 .await
607 .unwrap();
608
609 // 7. .. but abort the transaction and roll back.
610 pool_2.rollback(permit_2).await.unwrap();
611
612 // The value should still be the same as before.
613 let result = pool_2
614 .execute(async |pool| {
615 let row: (i64,) = query_as("SELECT x FROM test").fetch_one(pool).await?;
616 Ok(row.0)
617 })
618 .await
619 .unwrap();
620 assert_eq!(result, 5);
621 });
622
623 // 2. Pool 1 changes the value.
624 pool_1
625 .tx(async |tx| {
626 query("INSERT INTO test (x) VALUES (5)")
627 .execute(&mut **tx)
628 .await?;
629 Ok(())
630 })
631 .await
632 .unwrap();
633
634 // 3. Result is already 5 during "dirty read".
635 let result = pool_1
636 .tx(async |tx| {
637 let row: (i64,) = query_as("SELECT x FROM test").fetch_one(&mut **tx).await?;
638 Ok(row.0)
639 })
640 .await
641 .unwrap();
642 assert_eq!(result, 5);
643
644 // 4. Commit the change to database and free permit. This will allow now pool_2 to read the
645 // changed value.
646 pool_1.commit(permit_1).await.unwrap();
647
648 // Result is still 5 after commit.
649 let result = pool_1
650 .execute(async |pool| {
651 let row: (i64,) = query_as("SELECT x FROM test").fetch_one(pool).await?;
652 Ok(row.0)
653 })
654 .await
655 .unwrap();
656 assert_eq!(result, 5);
657
658 // Make sure we give pool 2 the time it needs to finish.
659 handle.await.unwrap();
660 }
661}