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
7mod enqueue;
8mod error;
9mod finalize;
10mod hydrate;
11mod import;
12mod lifecycle;
13mod migration;
14mod page;
15mod schema;
16
17pub use enqueue::enqueue;
18pub use error::{
19    ClaimError, EnqueueError, FinalizeError, ImportError, MutationError, PageError, SchemaError,
20    TransientKind,
21};
22pub use finalize::finalize_pending_delivery_for_migration;
23pub use import::import_for_migration;
24pub use lifecycle::{ack, claim, quarantine, release, renew, retry};
25pub use migration::{
26    CrateVersion, MIGRATIONS, Migration, MigrationCompatibility, MigrationCompatibilityError,
27    SCHEMA_VERSION,
28};
29pub use page::{SnapshotPager, begin_snapshot, page};
30pub use schema::check_schema;
31
32use dovecote::{EnqueueOutcome, FinalizeOutcome, ImportOutcome, ImportedDeliveryState, NewEvent};
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 [`enqueue`].
43pub async fn begin_write(pool: &SqlitePool) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
44    begin_write_with_config(pool, BusyConfig::default()).await
45}
46
47/// Alias for [`begin_write`] for callers about to enqueue an event.
48pub async fn begin_enqueue(
49    pool: &SqlitePool,
50) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
51    begin_write(pool).await
52}
53
54/// Bounded busy handling for SQLite's single-writer lock.
55#[derive(Clone, Copy, Debug, Eq, PartialEq)]
56pub struct BusyConfig {
57    timeout: Duration,
58    retries: u32,
59}
60
61impl BusyConfig {
62    /// Creates a policy with a finite per-lock wait and at most `retries`
63    /// immediate whole-operation retries. The total lock-wait budget is at
64    /// most `(retries + 1) * timeout` per operation.
65    pub const fn new(timeout: Duration, retries: u32) -> Self {
66        Self { timeout, retries }
67    }
68
69    /// Creates a policy using the default per-lock timeout.
70    pub const fn with_retries(retries: u32) -> Self {
71        Self::new(DEFAULT_BUSY_TIMEOUT, retries)
72    }
73
74    /// Maximum wait for one SQLite writer-lock acquisition.
75    pub const fn timeout(self) -> Duration {
76        self.timeout
77    }
78
79    /// Number of complete operation retries after the first lock wait.
80    pub const fn retries(self) -> u32 {
81        self.retries
82    }
83}
84
85impl Default for BusyConfig {
86    fn default() -> Self {
87        Self::new(DEFAULT_BUSY_TIMEOUT, DEFAULT_BUSY_RETRIES)
88    }
89}
90
91/// SQLite adapter for Dovecote's durable event and delivery schema.
92#[derive(Clone)]
93pub struct SqliteDovecote {
94    pool: SqlitePool,
95    busy: BusyConfig,
96}
97
98impl SqliteDovecote {
99    /// Creates an adapter with the documented bounded busy policy.
100    pub fn new(pool: SqlitePool) -> Self {
101        Self {
102            pool,
103            busy: BusyConfig::default(),
104        }
105    }
106
107    /// Creates an adapter with an explicit busy retry policy.
108    pub const fn with_busy_config(pool: SqlitePool, busy: BusyConfig) -> Self {
109        Self { pool, busy }
110    }
111
112    pub fn pool(&self) -> &SqlitePool {
113        &self.pool
114    }
115
116    pub const fn busy_config(&self) -> BusyConfig {
117        self.busy
118    }
119
120    /// Begins the caller transaction used for enqueue and application state.
121    /// It acquires SQLite's single writer slot before any adapter reads.
122    pub async fn begin_write(&self) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
123        begin_write_with_config(&self.pool, self.busy).await
124    }
125
126    /// Alias emphasizing that the returned transaction is suitable for
127    /// [`Self::enqueue`].
128    pub async fn begin_enqueue(&self) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
129        self.begin_write().await
130    }
131
132    pub async fn enqueue<'c>(
133        &self,
134        transaction: &mut Transaction<'c, Sqlite>,
135        event: NewEvent,
136    ) -> Result<EnqueueOutcome, EnqueueError> {
137        enqueue(transaction, event).await
138    }
139
140    /// Imports one already-validated event and its legacy delivery state in
141    /// the caller-owned transaction. This is migration infrastructure, not a
142    /// replacement for [`Self::enqueue`].
143    pub async fn import_for_migration<'c>(
144        &self,
145        transaction: &mut Transaction<'c, Sqlite>,
146        event: NewEvent,
147        state: ImportedDeliveryState,
148    ) -> Result<ImportOutcome, ImportError> {
149        import_for_migration(transaction, event, state).await
150    }
151
152    /// Records the legacy publisher's authoritative delivery time for a
153    /// canonical pending migration import. This operation is migration
154    /// infrastructure, not an ordinary acknowledgement shortcut.
155    pub async fn finalize_pending_delivery_for_migration<'c>(
156        &self,
157        transaction: &mut Transaction<'c, Sqlite>,
158        row_id: dovecote::RowId,
159        delivered_at: time::OffsetDateTime,
160    ) -> Result<FinalizeOutcome, FinalizeError> {
161        finalize_pending_delivery_for_migration(transaction, row_id, delivered_at).await
162    }
163
164    pub async fn check_schema(&self) -> Result<(), SchemaError> {
165        check_schema(&self.pool).await
166    }
167
168    pub async fn page(
169        &self,
170        after_row_id: Option<dovecote::RowId>,
171        limit: dovecote::Limit,
172    ) -> Result<Vec<dovecote::PagedEvent>, PageError> {
173        page(&self.pool, after_row_id, limit).await
174    }
175
176    pub async fn begin_snapshot(&self) -> Result<SnapshotPager, PageError> {
177        begin_snapshot(&self.pool).await
178    }
179
180    pub async fn claim(
181        &self,
182        worker: dovecote::WorkerId,
183        lease_for: dovecote::Lease,
184        limit: dovecote::Limit,
185    ) -> Result<Vec<dovecote::ClaimedEvent>, ClaimError> {
186        lifecycle::claim_with_config(&self.pool, worker, lease_for, limit, self.busy).await
187    }
188
189    pub async fn renew(
190        &self,
191        row_id: dovecote::RowId,
192        claim_token: &dovecote::ClaimToken,
193        lease_for: dovecote::Lease,
194    ) -> Result<(), MutationError> {
195        lifecycle::renew_with_config(&self.pool, row_id, claim_token, lease_for, self.busy).await
196    }
197
198    pub async fn ack(
199        &self,
200        row_id: dovecote::RowId,
201        claim_token: &dovecote::ClaimToken,
202    ) -> Result<(), MutationError> {
203        lifecycle::ack_with_config(&self.pool, row_id, claim_token, self.busy).await
204    }
205
206    pub async fn retry(
207        &self,
208        row_id: dovecote::RowId,
209        claim_token: &dovecote::ClaimToken,
210        failure: &dovecote::Failure,
211        backoff: dovecote::Delay,
212    ) -> Result<(), MutationError> {
213        lifecycle::retry_with_config(&self.pool, row_id, claim_token, failure, backoff, self.busy)
214            .await
215    }
216
217    pub async fn release(
218        &self,
219        row_id: dovecote::RowId,
220        claim_token: &dovecote::ClaimToken,
221        delay: dovecote::Delay,
222    ) -> Result<(), MutationError> {
223        lifecycle::release_with_config(&self.pool, row_id, claim_token, delay, self.busy).await
224    }
225
226    pub async fn quarantine(
227        &self,
228        row_id: dovecote::RowId,
229        claim_token: &dovecote::ClaimToken,
230        reason: &dovecote::QuarantineReason,
231    ) -> Result<(), MutationError> {
232        lifecycle::quarantine_with_config(&self.pool, row_id, claim_token, reason, self.busy).await
233    }
234}
235
236/// Kept private so adapter operations cannot accidentally use a worker clock.
237pub(crate) fn checked_milliseconds(value: Duration) -> Result<i64, String> {
238    if !value.is_zero() && !value.subsec_nanos().is_multiple_of(1_000_000) {
239        return Err("duration must be an exact whole number of milliseconds".to_owned());
240    }
241    i64::try_from(value.as_millis()).map_err(|_| "duration exceeds SQLite integer range".to_owned())
242}
243
244pub(crate) fn checked_busy_timeout(value: Duration) -> Result<i64, String> {
245    let milliseconds = checked_milliseconds(value)?;
246    if milliseconds > i64::from(i32::MAX) {
247        return Err("busy timeout exceeds SQLite's signed 32-bit millisecond range".to_owned());
248    }
249    Ok(milliseconds)
250}
251
252/// Acquires a pool connection, installs the busy timeout on that actual
253/// connection, verifies it, and only then starts `BEGIN IMMEDIATE`.
254pub(crate) async fn begin_immediate(
255    pool: &SqlitePool,
256    busy: BusyConfig,
257    _operation: &'static str,
258) -> Result<Transaction<'static, Sqlite>, sqlx::Error> {
259    let mut connection = pool.acquire().await?;
260    install_foreign_keys(&mut connection).await?;
261    install_busy_timeout(&mut connection, busy).await?;
262    Transaction::begin(
263        sqlx::pool::MaybePoolConnection::PoolConnection(connection),
264        Some(AssertSqlSafe("BEGIN IMMEDIATE").into_sql_str()),
265    )
266    .await
267}
268
269async fn begin_write_with_config(
270    pool: &SqlitePool,
271    busy: BusyConfig,
272) -> Result<Transaction<'static, Sqlite>, EnqueueError> {
273    validate_busy_config(busy).map_err(|detail| EnqueueError::Configuration { detail })?;
274    let mut tries = 0;
275    loop {
276        match begin_immediate(pool, busy, "begin write transaction").await {
277            Ok(transaction) => return Ok(transaction),
278            Err(source) if error::is_busy(&source) && tries < busy.retries() => {
279                tries += 1;
280            }
281            Err(source) => return Err(EnqueueError::sql("begin write transaction", source)),
282        }
283    }
284}
285
286pub(crate) fn validate_busy_config(busy: BusyConfig) -> Result<(), String> {
287    checked_busy_timeout(busy.timeout()).map(|_| ())
288}
289
290pub(crate) async fn begin_read(
291    pool: &SqlitePool,
292) -> Result<Transaction<'static, Sqlite>, sqlx::Error> {
293    let mut connection = pool.acquire().await?;
294    install_foreign_keys(&mut connection).await?;
295    Transaction::begin(
296        sqlx::pool::MaybePoolConnection::PoolConnection(connection),
297        Some(AssertSqlSafe("BEGIN").into_sql_str()),
298    )
299    .await
300}
301
302/// Completes a transaction while retaining the transaction value on commit
303/// failure long enough to await its rollback. SQLx's consuming
304/// `Transaction::commit` can only schedule a best-effort rollback when the
305/// commit fails; this path closes the owned transaction synchronously from the
306/// adapter's async operation instead.
307pub(crate) async fn commit_transaction(
308    mut transaction: Transaction<'static, Sqlite>,
309) -> Result<(), sqlx::Error> {
310    use sqlx_core::transaction::TransactionManager;
311
312    let result =
313        <sqlx::sqlite::SqliteTransactionManager as TransactionManager>::commit(&mut *transaction)
314            .await;
315    if result.is_err() {
316        let _ = <sqlx::sqlite::SqliteTransactionManager as TransactionManager>::rollback(
317            &mut *transaction,
318        )
319        .await;
320    }
321    result
322}
323
324pub(crate) async fn install_foreign_keys(
325    connection: &mut sqlx::pool::PoolConnection<Sqlite>,
326) -> Result<(), sqlx::Error> {
327    query("PRAGMA foreign_keys = ON")
328        .execute(&mut **connection)
329        .await?;
330    let enabled: i64 = query_scalar("PRAGMA foreign_keys")
331        .fetch_one(&mut **connection)
332        .await?;
333    if enabled != 1 {
334        return Err(sqlx::Error::Protocol(
335            "SQLite foreign-key enforcement could not be enabled".to_owned(),
336        ));
337    }
338    Ok(())
339}
340
341pub(crate) async fn install_busy_timeout(
342    connection: &mut sqlx::pool::PoolConnection<Sqlite>,
343    busy: BusyConfig,
344) -> Result<(), sqlx::Error> {
345    let milliseconds = checked_busy_timeout(busy.timeout())
346        .map_err(|detail| sqlx::Error::Protocol(format!("invalid busy configuration: {detail}")))?;
347    let statement = AssertSqlSafe(format!("PRAGMA busy_timeout = {milliseconds}"));
348    query(statement).execute(&mut **connection).await?;
349    let installed: i64 = query_scalar("PRAGMA busy_timeout")
350        .fetch_one(&mut **connection)
351        .await?;
352    if installed != milliseconds {
353        return Err(sqlx::Error::Protocol(format!(
354            "SQLite busy timeout installation mismatch: requested {milliseconds}, installed {installed}"
355        )));
356    }
357    Ok(())
358}
359
360/// SQLite exposes transaction state through its C API, not SQL. This is a
361/// read-only inspection and does not alter the caller transaction.
362pub(crate) async fn transaction_is_write(
363    transaction: &mut Transaction<'_, Sqlite>,
364) -> Result<bool, sqlx::Error> {
365    let mut handle = transaction.lock_handle().await?;
366    let state = unsafe {
367        libsqlite3_sys::sqlite3_txn_state(handle.as_raw_handle().as_ptr(), std::ptr::null())
368    };
369    Ok(state == libsqlite3_sys::SQLITE_TXN_WRITE)
370}
dovecote_sqlx_sqlite/lib.rs

dovecote_sqlx_sqlite/
lib.rs
