dovecote_sqlx_sqlite/
lib.rs

1//! `SQLite` schema and `SQLx` boundary for Dovecote.
2//!
3//! `SQLite`'s single-writer model is a distinct support contract. Write and
4//! claim transactions therefore use explicit `BEGIN IMMEDIATE`; busy errors
5//! are retried only by the bounded policy configured on [`SqliteDovecote`].
6#![warn(missing_docs)]
7
8mod delivery_state;
9mod enqueue;
10mod error;
11mod finalize;
12mod hydrate;
13mod import;
14mod lifecycle;
15mod lifecycle_mutation;
16mod migration;
17mod page;
18mod schema;
19mod scope;
20
21pub use error::{
22    ClaimError, EnqueueError, FinalizeError, ImportError, MutationError, PageError, SchemaError,
23    TransientKind,
24};
25pub use migration::{
26    CrateVersion, LEGACY_MIGRATION, MIGRATIONS, Migration, MigrationCompatibility,
27    MigrationCompatibilityError, SCHEMA_VERSION, V1_TENANT_ACTIVATE_SQL, V1_TENANT_PREPARE_SQL,
28};
29pub use page::SnapshotPager;
30pub use schema::check_schema;
31pub use scope::{AdminDovecote, TenantDovecote};
32
33use sqlx::{AssertSqlSafe, SqlSafeStr, Sqlite, SqlitePool, Transaction, query, query_scalar};
34use std::time::Duration;
35
36/// Default number of whole-operation retries after each configured busy timeout.
37pub const DEFAULT_BUSY_RETRIES: u32 = 3;
38/// Default per-connection wait before returning `SQLITE_BUSY`.
39pub const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
40
41/// Begins a caller write transaction using the default busy policy. The
42/// returned transaction is safe to pass to [`TenantDovecote::enqueue`].
43///
44/// # Errors
45/// Returns an error for invalid busy configuration or if the `SQLite` writer
46/// reservation cannot be acquired within the configured busy budget.
47pub async fn begin_write(pool: &SqlitePool) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
48    begin_write_with_config(pool, BusyConfig::default()).await
49}
50
51/// Alias for [`begin_write`] for callers about to enqueue an event.
52///
53/// # Errors
54/// Returns an error for invalid busy configuration or if the `SQLite` writer
55/// reservation cannot be acquired within the configured busy budget.
56pub async fn begin_enqueue(
57    pool: &SqlitePool,
58) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
59    begin_write(pool).await
60}
61
62/// Bounded busy handling for `SQLite`'s single-writer lock.
63#[derive(Clone, Copy, Debug, Eq, PartialEq)]
64pub struct BusyConfig {
65    timeout: Duration,
66    retries: u32,
67}
68
69impl BusyConfig {
70    /// Creates a policy with a finite per-lock wait and at most `retries`
71    /// immediate whole-operation retries. The total lock-wait budget is at
72    /// most `(retries + 1) * timeout` per operation.
73    #[must_use]
74    pub const fn new(timeout: Duration, retries: u32) -> Self {
75        Self { timeout, retries }
76    }
77
78    /// Creates a policy using the default per-lock timeout.
79    #[must_use]
80    pub const fn with_retries(retries: u32) -> Self {
81        Self::new(DEFAULT_BUSY_TIMEOUT, retries)
82    }
83
84    /// Maximum wait for one `SQLite` writer-lock acquisition.
85    #[must_use]
86    pub const fn timeout(self) -> Duration {
87        self.timeout
88    }
89
90    /// Number of complete operation retries after the first lock wait.
91    #[must_use]
92    pub const fn retries(self) -> u32 {
93        self.retries
94    }
95}
96
97impl Default for BusyConfig {
98    fn default() -> Self {
99        Self::new(DEFAULT_BUSY_TIMEOUT, DEFAULT_BUSY_RETRIES)
100    }
101}
102
103/// `SQLite` adapter for Dovecote's durable event and delivery schema.
104#[derive(Clone)]
105pub struct SqliteDovecote {
106    pool: SqlitePool,
107    busy: BusyConfig,
108}
109
110impl SqliteDovecote {
111    /// Creates an adapter with the documented bounded busy policy.
112    #[must_use]
113    pub fn new(pool: SqlitePool) -> Self {
114        Self {
115            pool,
116            busy: BusyConfig::default(),
117        }
118    }
119
120    /// Creates an adapter with an explicit busy retry policy.
121    #[must_use]
122    pub const fn with_busy_config(pool: SqlitePool, busy: BusyConfig) -> Self {
123        Self { pool, busy }
124    }
125
126    /// Borrows the pool used by this adapter.
127    #[must_use]
128    pub const fn pool(&self) -> &SqlitePool {
129        &self.pool
130    }
131
132    /// Returns the bounded busy policy used by adapter operations.
133    #[must_use]
134    pub const fn busy_config(&self) -> BusyConfig {
135        self.busy
136    }
137
138    /// Begins the caller transaction used for enqueue and application state.
139    /// It acquires `SQLite`'s single writer slot before any adapter reads.
140    ///
141    /// # Errors
142    /// Returns an error for invalid busy configuration or if the `SQLite` writer
143    /// reservation cannot be acquired within the configured busy budget.
144    pub async fn begin_write(&self) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
145        begin_write_with_config(&self.pool, self.busy).await
146    }
147
148    /// Alias emphasizing that the returned transaction is suitable for
149    /// [`TenantDovecote::enqueue`].
150    ///
151    /// # Errors
152    /// Returns an error for invalid busy configuration or if the `SQLite` writer
153    /// reservation cannot be acquired within the configured busy budget.
154    pub async fn begin_enqueue(&self) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
155        self.begin_write().await
156    }
157
158    /// Creates an ordinary handle restricted to one validated tenant.
159    #[must_use]
160    pub fn for_tenant(&self, tenant_id: dovecote::TenantId) -> TenantDovecote {
161        TenantDovecote::new(self.pool.clone(), tenant_id, self.busy)
162    }
163
164    /// Creates the explicit administrative handle for all-tenant reads and named writes.
165    #[must_use]
166    pub fn admin(&self) -> AdminDovecote {
167        AdminDovecote::new(self.pool.clone(), self.busy)
168    }
169
170    /// Verifies that the pool's current `SQLite` schema satisfies Dovecote.
171    ///
172    /// # Errors
173    /// Returns an error for an unsupported backend, missing or incompatible
174    /// migration markers, tables, constraints or indexes, or failed catalog reads.
175    pub async fn check_schema(&self) -> Result<(), SchemaError> {
176        check_schema(&self.pool).await
177    }
178}
179
180/// Kept private so adapter operations cannot accidentally use a worker clock.
181pub(crate) fn checked_milliseconds(value: Duration) -> Result<i64, String> {
182    if !value.is_zero() && !value.subsec_nanos().is_multiple_of(1_000_000) {
183        return Err("duration must be an exact whole number of milliseconds".to_owned());
184    }
185    i64::try_from(value.as_millis()).map_err(|_| "duration exceeds SQLite integer range".to_owned())
186}
187
188pub(crate) fn checked_busy_timeout(value: Duration) -> Result<i64, String> {
189    let milliseconds = checked_milliseconds(value)?;
190    if milliseconds > i64::from(i32::MAX) {
191        return Err("busy timeout exceeds SQLite's signed 32-bit millisecond range".to_owned());
192    }
193    Ok(milliseconds)
194}
195
196/// Acquires a pool connection, installs the busy timeout on that actual
197/// connection, verifies it, and only then starts `BEGIN IMMEDIATE`.
198pub(crate) async fn begin_immediate(
199    pool: &SqlitePool,
200    busy: BusyConfig,
201    _operation: &'static str,
202) -> Result<Transaction<'static, Sqlite>, sqlx::Error> {
203    let mut connection = pool.acquire().await?;
204    install_foreign_keys(&mut connection).await?;
205    install_busy_timeout(&mut connection, busy).await?;
206    Transaction::begin(
207        sqlx::pool::MaybePoolConnection::PoolConnection(connection),
208        Some(AssertSqlSafe("BEGIN IMMEDIATE").into_sql_str()),
209    )
210    .await
211}
212
213async fn begin_write_with_config(
214    pool: &SqlitePool,
215    busy: BusyConfig,
216) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
217    validate_busy_config(busy).map_err(|detail| EnqueueError::Configuration { detail })?;
218    let mut tries = 0;
219    loop {
220        match begin_immediate(pool, busy, "begin write transaction").await {
221            Ok(transaction) => return Ok(transaction),
222            Err(source) if error::is_busy(&source) && tries < busy.retries() => {
223                tries += 1;
224            }
225            Err(source) => return Err(EnqueueError::sql("begin write transaction", source)),
226        }
227    }
228}
229
230pub(crate) fn validate_busy_config(busy: BusyConfig) -> Result<(), String> {
231    checked_busy_timeout(busy.timeout()).map(|_| ())
232}
233
234pub(crate) async fn begin_read(
235    pool: &SqlitePool,
236) -> Result<Transaction<'static, Sqlite>, sqlx::Error> {
237    let mut connection = pool.acquire().await?;
238    install_foreign_keys(&mut connection).await?;
239    Transaction::begin(
240        sqlx::pool::MaybePoolConnection::PoolConnection(connection),
241        Some(AssertSqlSafe("BEGIN").into_sql_str()),
242    )
243    .await
244}
245
246/// Completes a transaction while retaining the transaction value on commit
247/// failure long enough to await its rollback. `SQLx`'s consuming
248/// `Transaction::commit` can only schedule a best-effort rollback when the
249/// commit fails; this path closes the owned transaction synchronously from the
250/// adapter's async operation instead.
251pub(crate) async fn commit_transaction(
252    mut transaction: Transaction<'static, Sqlite>,
253) -> Result<(), sqlx::Error> {
254    use sqlx_core::transaction::TransactionManager;
255
256    let result =
257        <sqlx::sqlite::SqliteTransactionManager as TransactionManager>::commit(&mut *transaction)
258            .await;
259    if result.is_err() {
260        let _ = <sqlx::sqlite::SqliteTransactionManager as TransactionManager>::rollback(
261            &mut *transaction,
262        )
263        .await;
264    }
265    result
266}
267
268pub(crate) async fn install_foreign_keys(
269    connection: &mut sqlx::pool::PoolConnection<Sqlite>,
270) -> Result<(), sqlx::Error> {
271    query("PRAGMA foreign_keys = ON")
272        .execute(&mut **connection)
273        .await?;
274    let enabled: i64 = query_scalar("PRAGMA foreign_keys")
275        .fetch_one(&mut **connection)
276        .await?;
277    if enabled != 1 {
278        return Err(sqlx::Error::Protocol(
279            "SQLite foreign-key enforcement could not be enabled".to_owned(),
280        ));
281    }
282    Ok(())
283}
284
285pub(crate) async fn install_busy_timeout(
286    connection: &mut sqlx::pool::PoolConnection<Sqlite>,
287    busy: BusyConfig,
288) -> Result<(), sqlx::Error> {
289    let milliseconds = checked_busy_timeout(busy.timeout())
290        .map_err(|detail| sqlx::Error::Protocol(format!("invalid busy configuration: {detail}")))?;
291    let statement = AssertSqlSafe(format!("PRAGMA busy_timeout = {milliseconds}"));
292    query(statement).execute(&mut **connection).await?;
293    let installed: i64 = query_scalar("PRAGMA busy_timeout")
294        .fetch_one(&mut **connection)
295        .await?;
296    if installed != milliseconds {
297        return Err(sqlx::Error::Protocol(format!(
298            "SQLite busy timeout installation mismatch: requested {milliseconds}, installed {installed}"
299        )));
300    }
301    Ok(())
302}
303
304/// `SQLite` exposes transaction state through its C API, not SQL. This is a
305/// read-only inspection and does not alter the caller transaction.
306pub(crate) async fn transaction_is_write(
307    transaction: &mut Transaction<'_, Sqlite>,
308) -> Result<bool, sqlx::Error> {
309    let mut handle = transaction.lock_handle().await?;
310    // SAFETY: SQLx's locked handle keeps the live connection exclusively borrowed
311    // until this call returns, preventing concurrent worker access. The pointer
312    // comes from that guard, and SQLite accepts a null schema name to inspect
313    // the transaction state across attached databases. This read changes no state.
314    let state = unsafe {
315        libsqlite3_sys::sqlite3_txn_state(handle.as_raw_handle().as_ptr(), std::ptr::null())
316    };
317    Ok(state == libsqlite3_sys::SQLITE_TXN_WRITE)
318}
dovecote_sqlx_sqlite/lib.rs

dovecote_sqlx_sqlite/
lib.rs
