Struct ockam::SqlxDatabase

source ·

pub struct SqlxDatabase {
    pub pool: Pool<Sqlite>,
}

Expand description

We use sqlx as our primary interface for interacting with the database The database driver is currently Sqlite

Fields§

§pool: Pool<Sqlite>

Pool of connections to the database

Implementations§

source §

impl SqlxDatabase

source

pub async fn create(path: impl AsRef<Path>) -> Result<SqlxDatabase, Error>

Constructor for a database persisted on disk

source

pub async fn in_memory(usage: &str) -> Result<Arc<SqlxDatabase>, Error>

Constructor for an in-memory database The implementation blocks during the creation of the database so that we don’t have to propagate async in all the code base when using an in-memory database, especially when writing examples

source

pub fn map_sql_err(err: Error) -> Error

Map a sqlx error into an ockam error

source

pub fn map_migrate_err(err: MigrateError) -> Error

Map a sqlx migration error into an ockam error

source

pub fn map_decode_err(err: Error) -> Error

Map a minicbor decode error into an ockam error

Methods from Deref<Target = Pool<Sqlite>>§

pub fn acquire( &self ) -> impl Future<Output = Result<PoolConnection<DB>, Error>> + 'static

Retrieves a connection from the pool.

The total time this method is allowed to execute is capped by [PoolOptions::acquire_timeout]. If that timeout elapses, this will return [Error::PoolClosed].

Note: Cancellation/Timeout May Drop Connections

If acquire is cancelled or times out after it acquires a connection from the idle queue or opens a new one, it will drop that connection because we don’t want to assume it is safe to return to the pool, and testing it to see if it’s safe to release could introduce subtle bugs if not implemented correctly. To avoid that entirely, we’ve decided to not gracefully handle cancellation here.

However, if your workload is sensitive to dropped connections such as using an in-memory SQLite database with a pool size of 1, you can pretty easily ensure that a cancelled acquire() call will never drop connections by tweaking your [PoolOptions]:

Set [test_before_acquire(false)][PoolOptions::test_before_acquire]
Never set [before_acquire][PoolOptions::before_acquire] or [after_connect][PoolOptions::after_connect].

This should eliminate any potential .await points between acquiring a connection and returning it.

pub fn try_acquire(&self) -> Option<PoolConnection<DB>>

Attempts to retrieve a connection from the pool if there is one available.

Returns None immediately if there are no idle connections available in the pool or there are tasks waiting for a connection which have yet to wake.

pub async fn begin(&self) -> Result<Transaction<'static, DB>, Error>

Retrieves a connection and immediately begins a new transaction.

pub async fn try_begin(&self) -> Result<Option<Transaction<'static, DB>>, Error>

Attempts to retrieve a connection and immediately begins a new transaction if successful.

pub fn close(&self) -> impl Future<Output = ()>

Shut down the connection pool, immediately waking all tasks waiting for a connection.

Upon calling this method, any currently waiting or subsequent calls to [Pool::acquire] and the like will immediately return [Error::PoolClosed] and no new connections will be opened. Checked-out connections are unaffected, but will be gracefully closed on-drop rather than being returned to the pool.

Returns a Future which can be .awaited to ensure all connections are gracefully closed. It will first close any idle connections currently waiting in the pool, then wait for all checked-out connections to be returned or closed.

Waiting for connections to be gracefully closed is optional, but will allow the database server to clean up the resources sooner rather than later. This is especially important for tests that create a new pool every time, otherwise you may see errors about connection limits being exhausted even when running tests in a single thread.

If the returned Future is not run to completion, any remaining connections will be dropped when the last handle for the given pool instance is dropped, which could happen in a task spawned by Pool internally and so may be unpredictable otherwise.

.close() may be safely called and .awaited on multiple handles concurrently.

pub fn is_closed(&self) -> bool

Returns true if [.close()][Pool::close] has been called on the pool, false otherwise.

pub fn close_event(&self) -> CloseEvent ⓘ

Get a future that resolves when [Pool::close()] is called.

If the pool is already closed, the future resolves immediately.

This can be used to cancel long-running operations that hold onto a [PoolConnection] so they don’t prevent the pool from closing (which would otherwise wait until all connections are returned).

Examples

These examples use Postgres and Tokio, but should suffice to demonstrate the concept.

Do something when the pool is closed:

use sqlx::PgPool;

let pool = PgPool::connect("postgresql://...").await?;

let pool2 = pool.clone();

tokio::spawn(async move {
    // Demonstrates that `CloseEvent` is itself a `Future` you can wait on.
    // This lets you implement any kind of on-close event that you like.
    pool2.close_event().await;

    println!("Pool is closing!");

    // Imagine maybe recording application statistics or logging a report, etc.
});

// The rest of the application executes normally...

// Close the pool before the application exits...
pool.close().await;

Cancel a long-running operation:

use sqlx::{Executor, PgPool};

let pool = PgPool::connect("postgresql://...").await?;

let pool2 = pool.clone();

tokio::spawn(async move {
    pool2.close_event().do_until(async {
        // This statement normally won't return for 30 days!
        // (Assuming the connection doesn't time out first, of course.)
        pool2.execute("SELECT pg_sleep('30 days')").await;

        // If the pool is closed before the statement completes, this won't be printed.
        // This is because `.do_until()` cancels the future it's given if the
        // pool is closed first.
        println!("Waited!");
    }).await;
});

// This normally wouldn't return until the above statement completed and the connection
// was returned to the pool. However, thanks to `.do_until()`, the operation was
// cancelled as soon as we called `.close().await`.
pool.close().await;

pub fn size(&self) -> u32

Returns the number of connections currently active. This includes idle connections.

pub fn num_idle(&self) -> usize

Returns the number of connections active and idle (not in use).

As of 0.6.0, this has been fixed to use a separate atomic counter and so should be fine to call even at high load.

This previously called [crossbeam::queue::ArrayQueue::len()] which waits for the head and tail pointers to be in a consistent state, which may never happen at high levels of churn.