sqlsrv 0.14.0

#![allow(clippy::doc_markdown)]

//! A library for implementing an in-process SQLite database server.
//!
//! # Connection pooling
//! sqlsrv implements connection pooling that reflects the concurrency model
//! of SQLite:  It supports multiple parallel readers, but only one writer.
//!
//! # Thread pooling
//! In addition to pooling connections, the library supports optionally using
//! a thread pool for diaptching database operations onto threads.
//!
//! # Features
//! | Feature  | Function
//! |----------|----------
//! | `tpool`  | Enable functions/methods that use a thread pool.

#![cfg_attr(docsrs, feature(doc_cfg))]

mod changehook;
mod err;
mod rawhook;
mod wrconn;

pub mod autovacuum;
pub mod utils;

use std::{fmt, mem::ManuallyDrop, path::Path, str::FromStr, sync::Arc};

use parking_lot::{Condvar, Mutex, MutexGuard};

use r2d2::{CustomizeConnection, PooledConnection};

pub use {r2d2, r2d2_sqlite::SqliteConnectionManager, rusqlite};

use rusqlite::{Connection, OpenFlags, params};

#[cfg(feature = "tpool")]
use threadpool::ThreadPool;

pub use changehook::ChangeLogHook;
pub use err::Error;
pub use rawhook::{Action, Hook};
pub use wrconn::WrConn;


/// Wrapper around a SQL functions registration callback used to select which
/// connection types to perform registrations on.
pub enum RegOn<F>
where
  F: Fn(&Connection) -> Result<(), rusqlite::Error>
{
  /// This registration callback should only be called for read-only
  /// connections.
  RO(F),

  /// This registration callback should only be called for the read/write
  /// connections.
  RW(F),

  /// This registration callback should be called for both the read-only and
  /// read/write connections.
  Both(F)
}


type RegCb = dyn Fn(&Connection) -> Result<(), rusqlite::Error> + Send + Sync;

enum CbType {
  Ro(Box<RegCb>),
  Rw(Box<RegCb>),
  Both(Box<RegCb>)
}


/// Used to register application callbacks to set up database schema.
pub trait SchemaMgr {
  /// Called just after the writer connection has been created and is intended
  /// to perform database initialization (create tables, add predefined rows,
  /// etc).
  ///
  /// `newdb` will be `true` if the database file did not exist prior to
  /// initialization.
  ///
  /// While this method can be used to perform schema upgrades, there are two
  /// specialized methods (`need_upgrade()` and `upgrade()`) that can be used
  /// for this purpose instead.
  ///
  /// The default implementation does nothing but returns `Ok(())`.
  ///
  /// # Errors
  /// Application-specific error.
  #[allow(unused_variables)]
  fn init(&self, conn: &mut Connection, newdb: bool) -> Result<(), Error> {
    Ok(())
  }

  /// Application callback used to determine if the database schema is out of
  /// date and needs to be updated.
  ///
  /// The default implementation does nothing but returns `Ok(false)`.
  ///
  /// # Errors
  /// Application-specific error.
  #[allow(unused_variables)]
  fn need_upgrade(&self, conn: &Connection) -> Result<bool, Error> {
    Ok(false)
  }

  /// Upgrade the database schema.
  ///
  /// This is called if [`SchemaMgr::need_upgrade()`] returns `Ok(true)`.
  ///
  /// The default implementation does nothing but returns `Ok(())`.
  ///
  /// # Errors
  /// Application-specific error.
  #[allow(unused_variables)]
  fn upgrade(&self, conn: &mut Connection) -> Result<(), Error> {
    Ok(())
  }
}


/// Read-only connection.
struct RoConn {
  regfuncs: Vec<CbType>
}

impl fmt::Debug for RoConn {
  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    write!(f, "RoConn {{}}")
  }
}


impl CustomizeConnection<rusqlite::Connection, rusqlite::Error> for RoConn {
  fn on_acquire(
    &self,
    conn: &mut rusqlite::Connection
  ) -> Result<(), rusqlite::Error> {
    conn.pragma_update(None, "foreign_keys", "ON")?;

    for rf in &self.regfuncs {
      match rf {
        CbType::Ro(f) | CbType::Both(f) => {
          f(conn)?;
        }
        CbType::Rw(_) => {}
      }
    }

    Ok(())
  }

  fn on_release(&self, _conn: rusqlite::Connection) {}
}


/// Builder for constructing a [`ConnPool`] object.
pub struct Builder {
  schmgr: Box<dyn SchemaMgr>,
  full_vacuum: bool,
  autovacuum: bool,
  max_readers: usize,
  hook: Option<Arc<dyn Hook + Send + Sync>>,
  regfuncs: Option<Vec<CbType>>,
  #[cfg(feature = "tpool")]
  tpool: Option<Arc<ThreadPool>>
}

/// Internal methods.
impl Builder {
  /// Open the writer connection.
  fn open_writer(&self, fname: &Path) -> Result<Connection, rusqlite::Error> {
    let conn = Connection::open(fname)?;
    conn.pragma_update(None, "journal_mode", "WAL")?;
    conn.pragma_update(None, "foreign_keys", "ON")?;

    // ToDo: Only enable incremental auto vacuum if autovacuum has been
    //       requested
    if self.autovacuum {
      conn.pragma_update(None, "auto_vacuum", "INCREMENTAL")?;
    }

    Ok(conn)
  }

  /// Run a full vacuum.
  ///
  /// This is an internal function that may be called by `build()` if a full
  /// vacuum has been requested.
  fn full_vacuum(conn: &Connection) -> Result<(), rusqlite::Error> {
    conn.execute("VACUUM;", params![])?;
    Ok(())
  }

  fn create_ro_pool(
    &self,
    fname: &Path,
    regfuncs: Vec<CbType>
  ) -> Result<r2d2::Pool<SqliteConnectionManager>, r2d2::Error> {
    let fl =
      OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX;
    let manager = SqliteConnectionManager::file(fname).with_flags(fl);
    let roconn_initterm = RoConn { regfuncs };
    let max_readers = u32::try_from(self.max_readers).unwrap();
    r2d2::Pool::builder()
      .max_size(max_readers)
      .connection_customizer(Box::new(roconn_initterm))
      .build(manager)
  }
}


impl Builder {
  /// Create a new `Builder` for constructing a [`ConnPool`] object.
  ///
  /// Default to not run a full vacuum of the database on initialization and
  /// create 2 read-only connections for the pool.
  /// No workers thread pool will be used.
  #[must_use]
  pub fn new(schmgr: Box<dyn SchemaMgr>) -> Self {
    Self {
      schmgr,
      full_vacuum: false,
      autovacuum: false,
      max_readers: 2,
      hook: None,
      regfuncs: None,
      #[cfg(feature = "tpool")]
      tpool: None
    }
  }

  /// Trigger a full vacuum when initializing the connection pool.
  ///
  /// Operates on an owned `Builder` object.
  #[must_use]
  pub const fn init_vacuum(mut self) -> Self {
    self.full_vacuum = true;
    self
  }

  /// Trigger a full vacuum when initializing the connection pool.
  ///
  /// Operates on a borrowed `Builder` object.
  pub const fn init_vacuum_r(&mut self) -> &mut Self {
    self.full_vacuum = true;
    self
  }

  /// Initialize the database in incremental vacuum mode.
  ///
  /// This will cause the writer's initialization will set the `auto_vacuum`
  /// pragma to `INCREMENTAL`.
  #[must_use]
  pub const fn autovacuum(mut self) -> Self {
    self.autovacuum = true;
    self
  }

  /// Initialize the database in incremental vacuum mode.
  ///
  /// This will cause the writer's initialization will set the `auto_vacuum`
  /// pragma to `INCREMENTAL`.
  pub const fn autovacuum_r(&mut self) -> &mut Self {
    self.autovacuum = true;
    self
  }

  /// Set maximum number of readers in the connection pool.
  ///
  /// Operates on an owned `Builder` object.
  #[must_use]
  pub const fn max_readers(mut self, n: usize) -> Self {
    self.max_readers = n;
    self
  }

  /// Set maximum number of readers in the connection pool.
  ///
  /// Operates on a borrowed `Builder` object.
  pub const fn max_readers_r(&mut self, n: usize) -> &mut Self {
    self.max_readers = n;
    self
  }

  /// Request that a "raw" update hook be added to the writer connection.
  ///
  /// Operates on an owned `Builder` object.
  #[must_use]
  pub fn hook(mut self, hook: Arc<dyn Hook + Send + Sync>) -> Self {
    self.hook = Some(hook);
    self
  }

  /// Request that a "raw" update hook be added to the writer connection.
  ///
  /// Operates on a borrowed `Builder` object.
  pub fn hook_r(&mut self, hook: Arc<dyn Hook + Send + Sync>) -> &mut Self {
    self.hook = Some(hook);
    self
  }

  /// Add a callback to register one or more scalar SQL functions.
  ///
  /// The closure should be wrapped in a `RegOn::RO()` if the function should
  /// only be registered on read-only connections.  `RegOn::RW()` is used to
  /// register the function on the read/write connection.  Use `RegOn::Both()`
  /// to register in both read-only and the read/write connection.
  #[must_use]
  pub fn reg_scalar_fn<F>(mut self, r: RegOn<F>) -> Self
  where
    F: Fn(&Connection) -> Result<(), rusqlite::Error> + Send + Sync + 'static
  {
    self.reg_scalar_fn_r(r);
    self
  }

  /// Add a callback to register one or more scalar SQL functions.
  ///
  /// This is the same as [`Builder::reg_scalar_fn()`], but it operates on
  /// `&mut Builder` rather than passing ownership.
  pub fn reg_scalar_fn_r<F>(&mut self, r: RegOn<F>) -> &mut Self
  where
    F: Fn(&Connection) -> Result<(), rusqlite::Error> + Send + Sync + 'static
  {
    match r {
      RegOn::RO(f) => {
        self
          .regfuncs
          .get_or_insert(Vec::new())
          .push(CbType::Ro(Box::new(f)));
      }
      RegOn::RW(f) => {
        self
          .regfuncs
          .get_or_insert(Vec::new())
          .push(CbType::Rw(Box::new(f)));
      }
      RegOn::Both(f) => {
        self
          .regfuncs
          .get_or_insert(Vec::new())
          .push(CbType::Both(Box::new(f)));
      }
    }
    self
  }

  #[cfg(feature = "tpool")]
  #[must_use]
  pub fn thread_pool(mut self, tpool: Arc<ThreadPool>) -> Self {
    self.tpool = Some(tpool);
    self
  }

  #[cfg(feature = "tpool")]
  pub fn thread_pool_r(&mut self, tpool: Arc<ThreadPool>) -> &mut Self {
    self.tpool = Some(tpool);
    self
  }

  /// Construct a connection pool.
  ///
  /// # Errors
  /// [`Error::Sqlite`] will be returned if a database error occurred.
  pub fn build<P>(mut self, fname: P) -> Result<ConnPool, Error>
  where
    P: AsRef<Path>
  {
    // ToDo: Use std::path::absolute() once stabilized
    let fname = fname.as_ref();
    let db_exists = fname.exists();

    //
    // Set up the read/write connection
    //
    // This must be done before creating the read-only connection pool, because
    // at that point the database file must already exist.
    //
    let mut conn = self.open_writer(fname)?;

    //
    // Register read/write connection functions
    //

    // Option<Vec<T>>  -->  Vec<T>
    let regfuncs = self.regfuncs.take().unwrap_or_default();

    // Call SQL function registration callbacks for read/write connection.
    for rf in &regfuncs {
      match rf {
        CbType::Rw(f) | CbType::Both(f) => {
          f(&conn)?;
        }
        CbType::Ro(_) => {}
      }
    }

    //
    // Perform schema initialization.
    //
    // This must be done after auto_vacuum is set, because auto_vacuum requires
    // configuration before any tables have been created.
    // See: https://www.sqlite.org/pragma.html#pragma_auto_vacuum
    //
    self.schmgr.init(&mut conn, !db_exists)?;
    if self.schmgr.need_upgrade(&conn)? {
      self.schmgr.upgrade(&mut conn)?;
    }

    //
    // Perform a full vacuum if requested to do so.
    //
    if self.full_vacuum {
      Self::full_vacuum(&conn)?;
    }

    //
    // Register a callback hook
    //
    if let Some(ref hook) = self.hook {
      rawhook::hook(&conn, hook)?;
    }

    //
    // Set up connection pool for read-only connections.
    //
    let rpool = self.create_ro_pool(fname, regfuncs)?;

    //
    // Prepare shared data
    //
    let iconn = InnerWrConn { conn, dirt: 0 };
    let inner = Inner { conn: Some(iconn) };
    let sh = Arc::new(Shared {
      inner: Mutex::new(inner),
      signal: Condvar::new()
    });

    Ok(ConnPool {
      rpool,
      sh,
      #[cfg(feature = "tpool")]
      tpool: self.tpool
    })
  }


  /// Construct a connection pool.
  ///
  /// Same as [`Builder::build()`], but register a change log callback on the
  /// writer as well.
  ///
  /// This method should not be called if the application has requested to add
  /// a raw update hook.
  ///
  /// # Errors
  /// [`Error::Sqlite`] is returned for database errors.
  ///
  /// # Panics
  /// This method will panic if a hook has been added to the Builder.
  pub fn build_with_changelog_hook<P, D, T>(
    mut self,
    fname: P,
    hook: Box<dyn ChangeLogHook<Database = D, Table = T> + Send>
  ) -> Result<ConnPool, Error>
  where
    P: AsRef<Path>,
    D: FromStr + Send + Sized + 'static,
    T: FromStr + Send + Sized + 'static
  {
    assert!(
      self.hook.is_some(),
      "Can't build a connection pool with both a raw and changelog hook"
    );

    // ToDo: Use std::path::absolute() once stabilized
    let fname = fname.as_ref();
    let db_exists = fname.exists();

    //
    // Set up the read/write connection
    //
    // This must be done before creating the read-only connection pool, because
    // at that point the database file must already exist.
    //
    let mut conn = self.open_writer(fname)?;

    //
    // Register read/write connection functions
    //

    // Option<Vec<T>>  -->  Vec<T>
    let regfuncs = self.regfuncs.take().unwrap_or_default();

    // Call SQL function registration callbacks for read/write connection.
    for rf in &regfuncs {
      match rf {
        CbType::Rw(f) | CbType::Both(f) => {
          f(&conn)?;
        }
        CbType::Ro(_) => {}
      }
    }


    //
    // Perform schema initialization.
    //
    // This must be done after auto_vacuum is set, because auto_vacuum requires
    // configuration before any tables have been created.
    // See: https://www.sqlite.org/pragma.html#pragma_auto_vacuum
    //
    self.schmgr.init(&mut conn, !db_exists)?;
    if self.schmgr.need_upgrade(&conn)? {
      self.schmgr.upgrade(&mut conn)?;
    }

    //
    // Perform a full vacuum if requested to do so.
    //
    if self.full_vacuum {
      Self::full_vacuum(&conn)?;
    }

    //
    // Register a callback hook
    //
    changehook::hook(&conn, hook)?;

    //
    // Set up connection pool for read-only connections.
    //
    let rpool = self.create_ro_pool(fname, regfuncs)?;

    //
    // Prepare shared data
    //
    let iconn = InnerWrConn { conn, dirt: 0 };
    let inner = Inner { conn: Some(iconn) };
    let sh = Arc::new(Shared {
      inner: Mutex::new(inner),
      signal: Condvar::new()
    });

    Ok(ConnPool {
      rpool,
      sh,
      #[cfg(feature = "tpool")]
      tpool: self.tpool
    })
  }
}


/// Inner writer connection object.
///
/// When the writer is acquired from the connection pool it passes an instance
/// of this struct to the WrConn object.
struct InnerWrConn {
  /// The writer connection.
  conn: Connection,

  /// Amount of accumulated dirt.
  ///
  /// Only used when the autoclean feature is used.
  dirt: usize
}


struct Inner {
  /// The writer connection.
  ///
  /// This is `None` when a `WrConn` exists, and is set to `Some()` by
  /// `WrConn`'s Drop implementation.
  conn: Option<InnerWrConn>
}

struct Shared {
  inner: Mutex<Inner>,
  signal: Condvar
}

impl Shared {
  #[inline]
  fn lock(&self) -> MutexGuard<'_, Inner> {
    self.inner.lock()
  }

  /*
  #[inline]
  fn with_lock<F, R>(&self, f: F) -> R
  where
    F: FnOnce(&mut Inner) -> R
  {
    let mut rw = self.lock();
    f(&mut rw)
  }
  */

  #[inline]
  fn with_lock_guard<F, R>(&self, f: F) -> R
  where
    F: FnOnce(MutexGuard<'_, Inner>) -> R
  {
    let g = self.lock();
    f(g)
  }
}


/// SQLite connection pool.
///
/// This is a specialized connection pool that is defined specifically for
/// sqlite, and only allows a single writer but multiple readers.
// Note:  In Rust the drop order of struct fields is in the order of
//        declaration.  If the writer is dropped before the readers, sqlite
//        will not clean up its wal files when the writet closes (presumably
//        because the readers are keeping the files locked).  Therefore it is
//        important that the r2d2 connection pool is declared before the
//        `Shared` buffer (since it contains the writer).
#[derive(Clone)]
pub struct ConnPool {
  rpool: r2d2::Pool<SqliteConnectionManager>,
  sh: Arc<Shared>,
  #[cfg(feature = "tpool")]
  tpool: Option<Arc<ThreadPool>>
}

impl ConnPool {
  /// Return the pool size.
  ///
  /// In effect, this is the size of the read-only pool plus one (for the
  /// read/write connection).
  #[must_use]
  pub fn size(&self) -> usize {
    (self.rpool.max_size() + 1) as usize
  }

  /// Acquire a read-only connection.
  ///
  /// # Errors
  /// [`r2d2::Error`] will be returned if a read-only connection could not be
  /// acquired.
  pub fn reader(
    &self
  ) -> Result<PooledConnection<SqliteConnectionManager>, r2d2::Error> {
    self.rpool.get()
  }

  /// Acquire the read/write connection.
  ///
  /// If the writer is already taken, then block and wait for it to become
  /// available.
  #[must_use]
  pub fn writer(&self) -> WrConn {
    let conn = self.sh.with_lock_guard(|mut g| {
      loop {
        if let Some(conn) = g.conn.take() {
          break conn;
        }
        self.sh.signal.wait(&mut g);
      }
    });

    WrConn {
      sh: Arc::clone(&self.sh),
      inner: ManuallyDrop::new(conn)
    }
  }

  /// Attempt to acquire the writer connection.
  ///
  /// Returns `Some(conn)` if the writer connection was available at the time
  /// of the request.  Returns `None` if the writer has already been taken.
  #[must_use]
  pub fn try_writer(&self) -> Option<WrConn> {
    let conn = self.sh.inner.lock().conn.take()?;
    Some(WrConn {
      sh: Arc::clone(&self.sh),
      inner: ManuallyDrop::new(conn)
    })
  }
}


/// Special queries.
impl ConnPool {
  /// Return the number of unused pages.
  ///
  /// # Errors
  /// - [`Error::R2D2`] indicates that it wasn't possible to acquire a
  ///   read-only connection from the connection pool.
  /// - [`Error::Sqlite`] means it was not possible to query the free page list
  ///   count.
  /// - [`Error::OutOfBound`] means that the free list count could not be
  ///   converted into a `usize`.
  pub fn freelist_count(&self) -> Result<usize, Error> {
    let npages = self.reader()?.query_row_and_then(
      "PRAGMA freelist_count;",
      [],
      |row| row.get::<_, i64>(0)
    )?;
    usize::try_from(npages).map_err(|_| {
      Error::oob("Freelist count could not be expressed as an `usize`")
    })
  }
}


/// Read-only connection processing.
impl ConnPool {
  /// Run a read-only database operation.
  ///
  /// # Errors
  /// The error type `E` is used to return application-defined errors, though
  /// it must be possible to convert a `r2d2::Error` into `E` using the `From`
  /// trait.
  #[inline]
  pub fn with_ro<T, F, E>(&self, f: F) -> Result<T, E>
  where
    T: Send + 'static,
    F: FnOnce(&Connection) -> Result<T, E> + Send + 'static,
    E: From<r2d2::Error>
  {
    // Acquire a read-only connection from the pool
    let conn = self.reader()?;

    // Run caller-provided closure.
    f(&conn)
  }

  /// Run a read-only database operation on a thread.
  ///
  /// # Errors
  /// [`r2d2::Error`] is returned if it wasn't possible to acquire a read-only
  /// connection from the connection pool.
  ///
  /// # Panics
  /// A thread pool must be associated with the [`ConnPool`] or this method
  /// will panic.
  #[cfg(feature = "tpool")]
  #[inline]
  pub fn with_ro_thrd<F>(&self, f: F) -> Result<(), r2d2::Error>
  where
    F: FnOnce(&Connection) + Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("ConnPool does to have a thread pool");
    };

    // Acquire a read-only connection from the pool and then run the provided
    // closure on a thread from the thread pool.
    let conn = self.reader()?;
    tpool.execute(move || {
      f(&conn);
    });
    Ok(())
  }

  /// Run a read-only database operation on a thread, allowing the caller to
  /// receive the `Result<T, E>` of the supplied closure using a
  /// one-shot channel.
  ///
  /// The supplied closure in `f` should return a `Result<T, E>` where the `Ok`
  /// case will be passed as a "set" value through the `swctx` channel, and the
  /// `Err` case will be passed as a "fail" value.
  ///
  /// # Errors
  /// [`r2d2::Error`] is returned if it wasn't possible to acquire a read-only
  /// connection from the connection pool.
  ///
  /// # Panics
  /// A thread pool must be associated with the [`ConnPool`] or this method
  /// will panic.
  #[cfg(feature = "tpool")]
  #[inline]
  pub fn with_ro_thrd_result<T, E, F>(
    &self,
    f: F
  ) -> Result<swctx::WaitCtx<T, (), E>, r2d2::Error>
  where
    T: Send + 'static,
    E: fmt::Debug + Send + 'static,
    F: FnOnce(&Connection) -> Result<T, E> + Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("ConnPool does to have a thread pool");
    };

    let conn = self.reader()?;

    let (sctx, wctx) = swctx::mkpair();

    // Ignore errors relating to pass the results back
    tpool.execute(move || match f(&conn) {
      Ok(t) => {
        let _ = sctx.set(t);
      }
      Err(e) => {
        let _ = sctx.fail(e);
      }
    });

    Ok(wctx)
  }
}

/// Read/Write connection processing.
impl ConnPool {
  /// Run a read/write database operation.
  ///
  /// # Errors
  /// Returns an application-specific type `E` on error.
  #[inline]
  pub fn with_rw<T, E, F>(&self, f: F) -> Result<T, E>
  where
    T: Send + 'static,
    E: fmt::Debug + Send + 'static,
    F: FnOnce(&mut WrConn) -> Result<T, E> + Send + 'static
  {
    let mut conn = self.writer();
    f(&mut conn)
  }

  /// Run a read/write database operation on a thread.
  ///
  /// The supplied closure should return an `Option<usize>`, where the `Some()`
  /// case denotes the specified amount of "dirt" should be added to the write
  /// connection.  `None` means no dirt should be added.
  ///
  /// # Panics
  /// A thread pool must be associated with the [`ConnPool`] or this method
  /// will panic.
  #[cfg(feature = "tpool")]
  #[inline]
  pub fn with_rw_thrd<F>(&self, f: F)
  where
    F: FnOnce(&mut WrConn) -> Option<usize> + Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("ConnPool does to have a thread pool");
    };

    let mut conn = self.writer();
    tpool.execute(move || {
      let dirt = f(&mut conn);
      if let Some(dirt) = dirt {
        conn.add_dirt(dirt);
      }
    });
  }

  /// Run a read/write database operation on a thread, allowing the
  /// caller to receive the `Result<T, E>` of the supplied closure using a
  /// one-shot channel.
  ///
  /// The supplied closure in `f` should return a `Result<T, E>` where the `Ok`
  /// case will be passed as a "set" value through the `swctx` channel, and the
  /// `Err` case will be passed as a "fail" value.
  ///
  /// # Panics
  /// A thread pool must be associated with the [`ConnPool`] or this method
  /// will panic.
  #[cfg(feature = "tpool")]
  #[inline]
  pub fn with_rw_thrd_result<T, E, F>(&self, f: F) -> swctx::WaitCtx<T, (), E>
  where
    T: Send + 'static,
    E: fmt::Debug + Send + 'static,
    F: FnOnce(&mut WrConn) -> Result<T, E> + Send + 'static
  {
    let Some(ref tpool) = self.tpool else {
      panic!("ConnPool does to have a thread pool");
    };

    let mut conn = self.writer();

    let (sctx, wctx) = swctx::mkpair();

    tpool.execute(move || match f(&mut conn) {
      Ok(t) => {
        let _ = sctx.set(t);
      }
      Err(e) => {
        let _ = sctx.fail(e);
      }
    });

    wctx
  }
}

/// Vacuuming methods.
impl ConnPool {
  #[cfg(feature = "tpool")]
  #[must_use]
  pub fn incremental_vacuum(
    &self,
    n: Option<usize>
  ) -> swctx::WaitCtx<(), (), Error> {
    self.with_rw_thrd_result(move |conn| conn.incremental_vacuum(n))
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :