Cargo.toml
README.md
www/index.md
www/changelog.md
src/err.rs
src/lib.rs
src/wrconn.rs
src/rawhook.rs
src/changehook.rs
src/utils.rs
examples/simple.rs

[package]
name = "sqlsrv"
version = "0.3.0"
edition = "2021"
license = "0BSD"
categories = [ "database" ]
keywords = [ "sqlite", "server" ]
repository = "https://repos.qrnch.tech/pub/sqlsrv"
description = "Utility functions for managing SQLite connections in a server application."
rust-version = "1.56"
exclude = [
  ".fossil-settings",
  ".efiles",
  ".fslckout",
  "build_docs.sh",
  "examples",
  "www",
  "rustfmt.toml"
]

[features]

tpool = ["dep:swctx", "dep:threadpool"]

[dependencies]
parking_lot = { version = "0.12.1" }
r2d2 = { version = "0.8.10" }
r2d2_sqlite = { version = "0.23.0" }
rusqlite = { version = "0.30.0", features = ["hooks"] }

#[cfg(feature = "tpool")]
use std::sync::Arc;

use rusqlite::{functions::FunctionFlags, params, Connection, ToSql};

use rand::Rng;

use sha2::{Digest, Sha256};

use sqlsrv::{RegOn, SchemaMgr};

    .incremental_autoclean(10, None)
    .reg_scalar_fn(RegOn::Both(register_genuid))
    .reg_scalar_fn(RegOn::RO(register_pwhash));

  // If 'tpool' is enabled, then create a thread pool with as many workers as
  // there are connections in the pool.
  #[cfg(feature = "tpool")]
  let tpool = {
    let tpool = Arc::new(threadpool::Builder::new().build());
    bldr.thread_pool_r(Arc::clone(&tpool));
    tpool
  };

  let connpool = bldr.build("test.sqlite").unwrap();

  // Acquire the writer connection and use it to conditionally create the
  // `stuff` table.
  {
    let wconn = connpool.writer();

      .query_row(params!("stuff"), |row| row.get::<usize, bool>(0))
      .unwrap();
    assert!(have);
  }

  #[cfg(feature = "tpool")]
  connpool
    .run_ro_thrd(|conn| {
      const SQL: &str = "SELECT * FROM snarks;";
      let mut _stmt = conn.prepare_cached(SQL).unwrap();
    })
    .unwrap();

  #[cfg(feature = "tpool")]
  connpool.run_rw_thrd(|conn| {
    const SQL: &str = "INSERT INTO whoop (name) VALUES (?);";
    let mut stmt = conn.prepare_cached(SQL).unwrap();
    stmt.execute(params!["test"]).unwrap();
    Some(1)
  });


  #[cfg(feature = "tpool")]
  connpool.run_rw_thrd(|conn| {
    const SQL: &str = "INSERT INTO uids (uid) VALUES (genuid(8));";
    let mut stmt = conn.prepare_cached(SQL).unwrap();
    stmt.execute(params![]).unwrap();
    Some(1)
  });

  #[cfg(feature = "tpool")]
  {
    let wctx: swctx::WaitCtx<_, _, rusqlite::Error> = connpool
      .run_rw_thrd_result(|conn| {
        const SQL: &str = "INSERT INTO uids (uid) VALUES (genuid(8));";
        let mut stmt = conn.prepare_cached(SQL)?;
        stmt.execute([])?;
        Ok(())
      });
    wctx.wait().unwrap();
  }


  #[cfg(feature = "tpool")]
  tpool.join();
}


/// Register a pwhash() SQL function which returns a hex-encoded version of
/// the SHA256 hash of the input string.
fn register_pwhash(db: &Connection) -> Result<(), rusqlite::Error> {
  db.create_scalar_function(

//! 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.
//!
//! # Incremental auto-clean
//! The connection pool has built-in support for setting up incremental
//! autovacuum, and can be configured to implicitly run incremental vacuuming.
//!
//! To use this feature, a "maximum dirt" value is configured on the connection
//! pool.  Whenever the writer connection performs changes to the database it
//! can add "dirt" to the connection.  When the writer connection is returned
//! to the connection pool it checks to see if the amount of dirt is equal to
//! or greater than the configured "maximum dirt" threshold.  If the threshold
//! has been reached, an incremental autovacuum is performed.
//!
//! # 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 utils;

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

use parking_lot::{Condvar, Mutex};

use r2d2::{CustomizeConnection, PooledConnection};

use r2d2_sqlite::SqliteConnectionManager;

pub use r2d2;
pub use rusqlite;

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

#[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> + Sync + Sync
{
  /// 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 {

    Ok(())
  }

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












/// Builder for constructing a [`ConnPool`] object.
pub struct Builder {
  schmgr: Box<dyn SchemaMgr>,
  full_vacuum: bool,
  max_readers: usize,


  autoclean: Option<AutoClean>,
  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)?;

    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.
  pub fn new(schmgr: Box<dyn SchemaMgr>) -> Self {
    Self {
      schmgr,
      full_vacuum: false,
      max_readers: 2,


      autoclean: None,
      hook: None,
      regfuncs: None,
      #[cfg(feature = "tpool")]
      tpool: None
    }
  }

  /// Trigger a full vacuum when initializing the connection pool.
  ///
  /// Operates on an owned `Builder` object.
  pub fn init_vacuum(mut self) -> Self {

  ///
  /// Operates on a borrowed `Builder` object.
  pub 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.
  pub fn hook(mut self, hook: Arc<dyn Hook + Send + Sync>) -> Self {
    self.hook = Some(hook);
    self
  }

          .get_or_insert(Vec::new())
          .push(CbType::Both(Box::new(f)));
      }
    }
    self
  }

  #[cfg(feature = "tpool")]
  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.
  pub fn build<P>(mut self, fname: P) -> Result<ConnPool, Error>
  where
    P: AsRef<Path>
  {
    // ToDo: Use std::path::absolute() once stabilized

    //
    // Register a callback hook
    //
    if let Some(ref hook) = self.hook {
      rawhook::hook(&conn, Arc::clone(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(),
      autoclean: self.autoclean.clone()
    });

    Ok(ConnPool {

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


  /// Construct a connection pool.
  ///
  /// Same as [`Builder::build()`], but register a change log callback on the

    }

    //
    // 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(),
      autoclean: self.autoclean.clone()
    });

    Ok(ConnPool {

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


/// Inner writer connection object.
///

  signal: Condvar,
  autoclean: Option<AutoClean>
}


/// SQLite connection pool.
///
/// This is a specialized connection pool that is defined specifically for
/// sqlite, and only allows a single writer but multiple readers.
pub struct ConnPool {
  sh: Arc<Shared>,
  rpool: r2d2::Pool<SqliteConnectionManager>,
  #[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).
  pub fn size(&self) -> usize {
    (self.rpool.max_size() + 1) as usize
  }

  /// Acquire a read-only connection.
  pub fn reader(
    &self
  ) -> Result<PooledConnection<SqliteConnectionManager>, r2d2::Error> {
    self.rpool.get()
  }

  /// 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.
  pub fn try_writer(&self) -> Option<WrConn> {
    let mut g = self.sh.inner.lock();
    let conn = g.conn.take()?;



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


/// Special queries.
impl ConnPool {
  /// Return the number of unused pages.
  pub fn freelist_count(&self) -> Result<usize, Error> {
    Ok(self.reader()?.query_row_and_then(
      "PRAGMA freelist_count;'",
      [],
      |row| row.get(0)
    )?)
  }
}


pub enum RunError<E> {
  R2D2(r2d2::Error),
  App(E)
}

/// Read-only connection processing.
impl ConnPool {
  /// Run a read-only database operation.
  ///
  /// # Errors
  /// If a connection could not be acquired from the connection pool,
  /// `Err(RunError::R2D2(r2d2::Error))` will be returned.  If the application
  /// callback fails, this function will return `Err(RunError::App(E))`.
  pub fn run_ro<T, E, F>(&self, f: F) -> Result<T, RunError<E>>
  where
    T: Send + 'static,
    E: fmt::Debug + Send + 'static,
    F: FnOnce(&Connection) -> Result<T, E> + Send + 'static
  {
    // Acquire a read-only connection from the pool
    let conn = self.reader().map_err(|e| RunError::R2D2(e))?;

    // Run caller-provided closure.  On error map error to RunError::App().
    f(&conn).map_err(|e| RunError::App(e))
  }

  /// Run a read-only database operation on a thread.
  ///
  /// # Panics
  /// A thread pool must be associated with the [`ConnPool`] or this method
  /// will panic.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn run_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.
  ///
  /// # Panics
  /// A thread pool must be associated with the [`ConnPool`] or this method
  /// will panic.
  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn run_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();

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

    Ok(wctx)
  }
}

/// Read/Write connection processing.
impl ConnPool {
  /// Run a read/write database operation.
  pub fn run_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")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn run_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")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn run_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) => sctx.set(t),
      Err(e) => sctx.fail(e)
    });

    wctx
  }
}



impl ConnPool {



  #[cfg(feature = "tpool")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
  pub fn incremental_vacuum(
    &self,
    n: Option<usize>
  ) -> swctx::WaitCtx<(), (), rusqlite::Error> {









    self.run_rw_thrd_result(move |conn| conn.incremental_vacuum(n))







  }
}

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

//! Utility functions around SQL commands.

use rusqlite::{params, Connection};

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

#[cfg(feature = "tpool")]
use super::ConnPool;

/// Return the number of pages in the freelist.
pub fn freelist_count(conn: &Connection) -> Result<usize, rusqlite::Error> {
  conn.query_row_and_then("PRAGMA freelist_count;'", [], |row| row.get(0))
}

/// Run an incremental vacuum.
///
/// If `n` is `None` the entrire list of free pages will be processed.  If it
/// is `Some(n)` then only up to `n` pages will be processed.
pub fn incremental_vacuum(
  conn: &Connection,
  n: Option<usize>
) -> Result<(), rusqlite::Error> {
  if let Some(n) = n {
    conn.execute("PRAGMA incremental_vacuum(?);", params![n])
  } else {
    conn.execute("PRAGMA incremental_vacuum;", params![])
  }
  .map(|_| ())
}

#[cfg(feature = "tpool")]
#[cfg_attr(docsrs, doc(cfg(feature = "tpool")))]
pub fn pooled_incremental_vacuum(
  cpool: &ConnPool,
  tpool: &ThreadPool,
  n: Option<usize>
) -> swctx::WaitCtx<(), (), rusqlite::Error> {
  let (sctx, wctx) = swctx::mkpair();

  let conn = cpool.writer();

  tpool.execute(move || match conn.incremental_vacuum(n) {
    Ok(_) => sctx.set(()),
    Err(e) => sctx.fail(e)
  });

  wctx
}

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

impl WrConn {
  /// Add dirt to the writer connection.
  pub fn add_dirt(&mut self, weight: usize) {
    self.inner.dirt = self.inner.dirt.saturating_add(weight);
  }
}

impl WrConn {
  pub fn incremental_vacuum(
    &self,
    n: Option<usize>
  ) -> Result<(), rusqlite::Error> {
    if let Some(n) = n {
      self
        .inner
        .conn
        .execute("PRAGMA incremental_vacuum(?);", params![n])
    } else {
      self
        .inner
        .conn
        .execute("PRAGMA incremental_vacuum;", params![])
    }
    .map(|_| ())
  }
}

impl Deref for WrConn {
  type Target = Connection;

  #[inline(always)]
  fn deref(&self) -> &Connection {
    &self.inner.conn

# Change Log

## [Unreleased]

[Details](/vdiff?from=sqlsrv-0.3.0&to=trunk)

### Added

### Changed

### Removed

---

## [0.3.0] - 2024-02-10

[Details](/vdiff?from=sqlsrv-0.2.0&to=sqlsrv-0.3.0)

### Added

- Re-export `r2d2`.
- Add `ConnPool::size()` for getting number of connections in connection pool.
- Add `ConnPool::freelist_count()` for getting number of unused pages in the
  database.
- `WrConn::incremental_vacuum()` added.
- Add an `utils` sudmodule for collecting SQL command wrappers.
- Add several new `ConnPool` wrappers for running closures with either
  read-only or read/write connections.

### Changed

- `tpool` feature is no longer the default.
- Redesigned thread pooling to support sharing a thread pool with others.

### Removed

- Removed support for the thread pool in the `ConnPool` and opt for providing
  functions/methods that take in a reference to a `threadpool::ThreadPool`
  instead.

---

## [0.2.0] - 2024-01-28

[Details](/vdiff?from=sqlsrv-0.1.1&to=sqlsrv-0.2.0)