# Async, pooling, and transactions
*Status: design, not yet implemented.*
## Motivation
Squire currently exposes a sync SQLite API. Connections are `Send` under the `multi-thread` feature and `Send + Sync` under `serialized`, but there is no async surface, no built-in connection pool, and no `Transaction` wrapper. Downstream applications — the animating example being an Axum server — need to run SQLite work from async tasks without inheriting the pitfalls of a naïve "spawn_blocking per call" model:
- **Priority inversion.** A long analytical query on a shared blocking pool blocks ahead of a fast interactive lookup.
- **Intra-process write contention.** SQLite allows a single writer. Multiple async tasks racing on write connections produce spurious `SQLITE_BUSY`.
- **Cancellation is fake.** Dropping a future that wraps `spawn_blocking` abandons the join handle; the SQLite call runs to completion.
- **Unbounded queueing.** No backpressure.
- **Repeated statement preparation.** The obvious "send a closure" model discards prepared statements between calls.
This design addresses each of these with **simple, composable building blocks** that route users to correct SQLite usage by construction.
## Design principles
- **The type system routes correctness.** Read and write paths are distinct types. You cannot issue a mutation through a reader; you cannot forget to acquire the writer slot.
- **Intra-process write contention is eliminated by construction.** A pool has at most one in-flight writer at a time. `SQLITE_BUSY` from our own connections cannot happen. External-process contention is handled by `busy_timeout` + bounded retry.
- **Cancellation is real.** Dropping an async call triggers `sqlite3_interrupt` on the in-flight statement.
- **Backpressure is the default.** Streams and mailboxes are bounded. No unbounded queue anywhere in the surface.
- **Composition over configuration.** Apps with heterogeneous workloads (interactive + analytical) compose multiple pools rather than tuning priority knobs on a single shared pool.
## Non-goals (deferred)
- Static SQL validation (sqlx-style). The trait shape leaves room; implementation is a separate project.
- Sibling-to-`.rs` SQL file discovery by query macros. Gated on unstable `proc_macro_span`.
- Runtime integrations beyond Tokio. `futures` / `async-std` support can be added later behind parallel feature flags.
- Priority lanes within a single connection's mailbox. Users compose pools instead.
- Ad-hoc SQL statement cache. Only `#[squire::query]` types are cached initially.
## Core abstractions
### `Query` and `Mutation` traits
Typed SQL operations implement one of two traits. The split is the type-level signal of read-vs-write intent; macros emit the appropriate trait based on SQL classification (`SELECT` / `WITH ... SELECT` → `Query`; everything else, including DDL and most PRAGMAs, → `Mutation`).
```rust
pub trait Query {
type Parameters<'s>: Parameters<'s>;
type Output<'c: 's, 's>;
/// Output type usable across async boundaries (owned, Send, 'static).
type OwnedOutput: Send + 'static;
fn prepare(connection: &Connection) -> Result<Statement<'_>>;
fn output<'c: 's, 's>(execution: Execution<'c, 's>) -> Result<Self::Output<'c, 's>>;
fn owned_output(execution: Execution<'_, '_>) -> Result<Self::OwnedOutput>;
}
pub trait Mutation {
type Parameters<'s>: Parameters<'s>;
type Output<'c: 's, 's>;
type OwnedOutput: Send + 'static;
fn prepare(connection: &Connection) -> Result<Statement<'_>>;
fn output<'c: 's, 's>(execution: Execution<'c, 's>) -> Result<Self::Output<'c, 's>>;
fn owned_output(execution: Execution<'_, '_>) -> Result<Self::OwnedOutput>;
}
```
- `Output<'c, 's>` preserves the current sync ergonomics (borrowed rows, zero-copy strings, etc.).
- `OwnedOutput` is the variant that crosses the actor mailbox. For most queries this is the "collected" form (e.g., `Vec<User>`, `User`, `()`). Macros emit both.
- Dropping the `&'static str` SQL assumption keeps the door open for native UTF-16 support.
### `Reader` and `Writer` traits (sync)
```rust
pub trait Reader {
fn run<Q: Query>(&self, params: Q::Parameters<'_>) -> Result<Q::Output<'_, '_>>;
}
pub trait Writer: Reader {
fn mutate<M: Mutation>(&self, params: M::Parameters<'_>) -> Result<M::Output<'_, '_>>;
}
```
Implemented by:
- `Connection` → `Writer`.
- `ReadTransaction<'c>` → `Reader` only.
- `WriteTransaction<'c>` → `Writer`.
- `PooledReader<'p>` → `Reader` only.
- `PooledWriter<'p>` → `Writer`.
- `Pool` → `Writer` (routes internally).
Generic code composes naturally:
```rust
fn load_user(db: &impl Reader, id: i64) -> Result<User> { ... }
fn create_user(db: &impl Writer, input: &NewUser) -> Result<RowId> { ... }
```
Ad-hoc SQL (`conn.prepare("..."); stmt.query(...)`) stays on the concrete types, off the trait. Transaction construction (`read_transaction`, `write_transaction`) is also on the concrete types because the return lifetime ties to `&mut self`.
### `AsyncReader` and `AsyncWriter` traits
```rust
pub trait AsyncReader {
fn run<Q>(&self, params: Q::Parameters<'static>)
-> impl Future<Output = Result<Q::OwnedOutput>> + Send
where
Q: Query + 'static,
Q::Parameters<'static>: Send + 'static;
fn stream<Q>(&self, params: Q::Parameters<'static>)
-> impl Stream<Item = Result<Q::OwnedOutput>> + Send
where
Q: Query + 'static,
Q::Parameters<'static>: Send + 'static;
fn query_sql<P>(&self, sql: &str, params: P)
-> impl Future<Output = Result<Rows>> + Send
where
P: for<'s> Parameters<'s> + Send + 'static;
fn with_reader<F, R>(&self, f: F) -> impl Future<Output = Result<R>> + Send
where
F: FnOnce(&mut Connection) -> Result<R> + Send + 'static,
R: Send + 'static;
}
pub trait AsyncWriter: AsyncReader {
fn mutate<M>(&self, params: M::Parameters<'static>)
-> impl Future<Output = Result<M::OwnedOutput>> + Send
where
M: Mutation + 'static,
M::Parameters<'static>: Send + 'static;
fn mutate_stream<M>(&self, params: M::Parameters<'static>)
-> impl Stream<Item = Result<M::OwnedOutput>> + Send
where
M: Mutation + 'static,
M::Parameters<'static>: Send + 'static;
fn execute_sql<P>(&self, sql: &str, params: P)
-> impl Future<Output = Result<RowId>> + Send
where
P: for<'s> Parameters<'s> + Send + 'static;
fn with_writer<F, R>(&self, f: F) -> impl Future<Output = Result<R>> + Send
where
F: FnOnce(&mut Connection) -> Result<R> + Send + 'static,
R: Send + 'static;
}
```
Implemented by:
- `AsyncConnection` → `AsyncWriter`.
- `AsyncPool` → `AsyncWriter` (routes internally between reader and writer actors).
- `AsyncReadTransaction<'p>` → `AsyncReader` only.
- `AsyncWriteTransaction<'p>` → `AsyncWriter`.
Notes on the signatures:
- `Parameters<'static>` on the async path expresses that anything crossing the mailbox must be owned. Macros generate a sensible `Parameters<'static>` impl for typed queries (the concrete `GetUser(i64)` type, for example, trivially satisfies this).
- `OwnedOutput` on the return type expresses the same constraint in the other direction.
- `stream` produces `Q::OwnedOutput` per item, not raw `Row`. For typical queries this means `User` per item (one row's owned form); for queries with richer aggregation the `owned_output` impl can collect multi-row groups. The default macro-emitted shape is "one row → one `OwnedOutput`."
- `query_sql` / `execute_sql` return coarse types (`Rows`, `RowId`) because they forgo the `Query`/`Mutation` type and thus the typed output. They are the ad-hoc escape.
- `with_reader` / `with_writer` take `&mut Connection` — full sync API power inside the closure, including `Statement`, sync `Transaction`, iterators. `Connection::close(mut self)` is not callable through `&mut`, so the closure cannot close the pooled connection. The closure cannot `.await` (it runs synchronously on the worker thread).
### Transaction types
```rust
pub struct ReadTransaction<'c> { /* inner: &'c mut Connection */ }
pub struct WriteTransaction<'c> { /* inner: &'c mut Connection */ }
impl<'c> ReadTransaction<'c> {
pub fn commit(self) -> Result<()>; // also: just drop (rollback is identical for reads)
}
impl<'c> WriteTransaction<'c> {
pub fn commit(self) -> Result<()>;
pub fn rollback(self) -> Result<()>;
}
impl<'c> Reader for ReadTransaction<'c> { /* ... */ }
impl<'c> Reader for WriteTransaction<'c> { /* ... */ }
impl<'c> Writer for WriteTransaction<'c> { /* ... */ }
```
- `ReadTransaction` opens with `BEGIN DEFERRED`. Read-only; obtains a consistent snapshot.
- `WriteTransaction` opens with `BEGIN IMMEDIATE` by default. This acquires the write lock up front, eliminating the upgrade-from-`DEFERRED` deadlock that is the single most common source of `SQLITE_BUSY`.
- `Drop` without `.commit()` triggers `ROLLBACK`. Drop is infallible; errors during rollback are logged (TBD: or surfaced via a callback).
- The absence of `Writer` on `ReadTransaction` is the correctness property: `read_tx.mutate::<M>(...)` does not compile.
Async counterparts `AsyncReadTransaction<'p>` / `AsyncWriteTransaction<'p>` have the same shape. They hold the acquired reader/writer slot for their lifetime; on drop, a rollback command is sent to the worker. Because `Drop` cannot `.await`, the rollback is fire-and-forget from the caller's perspective — the actor processes it before accepting further work.
### Pool types
```rust
pub struct PoolBuilder<L> {
database: Database<L>,
readers: usize, // default: num_cpus
writers: usize, // default: 1
busy_timeout: Duration, // default: 5s
wal: bool, // default: true
retry_policy: RetryPolicy,
// ...
}
pub struct Pool { /* sync */ }
pub struct AsyncPool { /* async, feature-gated */ }
impl Pool {
pub fn builder<L>(database: Database<L>) -> PoolBuilder<L>;
pub fn read(&self) -> Result<PooledReader<'_>>; // blocks on reader availability
pub fn write(&self) -> Result<PooledWriter<'_>>; // blocks on writer availability
}
impl Reader for Pool { /* routes via read() */ }
impl Writer for Pool { /* routes via write() */ }
impl AsyncPool {
pub fn builder<L>(database: Database<L>) -> PoolBuilder<L>;
pub async fn read(&self) -> Result<AsyncPooledReader<'_>>;
pub async fn write(&self) -> Result<AsyncPooledWriter<'_>>;
}
impl AsyncReader for AsyncPool { /* ... */ }
impl AsyncWriter for AsyncPool { /* ... */ }
```
- Default topology: **1 writer + N readers, WAL mode.** WAL is enabled by pool initialization; no user action required.
- Applications compose multiple pools for workload isolation (interactive + analytical). The pool shape is the contract; users build as many as they need.
- Sync pool internals: `Mutex<VecDeque<Connection>>` + `Condvar` for the reader pool; `Mutex<Option<Connection>>` (or a semaphore-of-1) for the writer. No external dependency.
- Async pool internals: one actor per connection. Reader slot = a set of N actors; writer slot = one actor. Acquire = pick a free reader / await the writer.
## Actor model (async)
Each `AsyncConnection` is backed by:
- **One dedicated OS thread** (`std::thread::spawn`, not Tokio's blocking pool).
- **A Tokio `mpsc` mailbox** for commands; each command carries a `oneshot` sender for the response.
- **A supervisor handle** holding the raw `*mut sqlite3` pointer for calling `sqlite3_interrupt` from other threads (used for drop-cancellation).
- **A per-connection statement cache** (see below).
Commands (illustrative):
```rust
enum Command {
RunTyped { run: Box<dyn FnOnce(&mut Worker) -> Box<dyn Any + Send> + Send> },
Stream { run: ..., tx: tokio::sync::mpsc::Sender<Result<Row>> },
WithConn { closure: Box<dyn FnOnce(&mut Connection) -> Box<dyn Any + Send> + Send> },
Transaction { ... },
Interrupt, // out-of-band, via a second channel or shared flag
Shutdown,
}
```
The exact wire format is an implementation detail; the public API does not expose `Command`.
### Cancellation flow
1. A future returned from `run` / `mutate` / `with_reader` / `with_writer` (or a `Stream` from `stream` / `mutate_stream`) is dropped.
2. The drop impl signals the actor's supervisor: "the call in flight on this mailbox slot should be interrupted."
3. The supervisor calls `sqlite3_interrupt(handle)` on the worker's `sqlite3*`. This is safe from any thread; it sets a flag that `sqlite3_step` checks.
4. The worker pops out of `step` with `SQLITE_INTERRUPT`, unwinds the command, and returns to the mailbox.
5. Subsequent commands proceed normally. No connection reset required.
For streams, the same flow applies: dropping the `Stream` drops its underlying mailbox future and triggers the interrupt.
### Statement cache
- Per-worker `HashMap<TypeId, Statement<'static>>` keyed by `TypeId::of::<Q>()`. The `Statement<'static>` is safe because the statement's lifetime parameter is the worker-owned `Connection`, which outlives the cache.
- On first `run::<Q>` / `mutate::<M>`, the worker calls `Q::prepare(conn)` and stores the result. Subsequent calls re-bind and re-execute.
- LRU eviction at a configurable capacity (default TBD — a few hundred).
- Ad-hoc SQL (`query_sql` / `execute_sql`) is not cached. A string-keyed LRU can be added later.
## Streaming semantics
- `stream` and `mutate_stream` use a bounded `tokio::sync::mpsc` channel between the worker and the consumer. Default bound: 8 items for `stream`, 32 for `mutate_stream`. Configurable per call.
- Backpressure is eager by default: if the consumer stalls, the worker pauses stepping. **This means the reader connection (or writer slot) is held for the stream's lifetime.** A slow consumer on a stream holds its slot; on `mutate_stream`, this blocks every other writer in the pool.
- Rationale for separate `mutate_stream` name: the name is the warning. `UPDATE ... RETURNING` streamed with a slow consumer is a real footgun, and making it a distinct method makes the implications visible at the call site.
- `stream` is available on pool types only, not on transaction types. A `Stream` that holds a transaction reference creates lifetime constraints that are painful to reconcile with the `'static + Send` bounds async tasks require. Inside a transaction, use `with_reader` / `with_writer` and iterate synchronously.
## Contention handling
SQLite's lock contention sources, and how each is addressed:
| Two of our writers racing | Writer slot (semaphore-of-1). Only one in-flight writer per pool. |
| Deferred→reserved upgrade deadlock | `BEGIN IMMEDIATE` default in `WriteTransaction`. |
| External-process writer | `busy_timeout` (5s default). Retries inside SQLite. |
| External-process persistent contention | Bounded retry policy on top of `busy_timeout` exhaustion. |
| Slow reader starving writers | Non-issue in WAL mode. |
| Slow writer starving readers | Non-issue in WAL mode. |
| Long analytical query blocking interactive | Separate pools (composition, not priority knobs). |
## Runtime coupling
- Async support lives behind an `async` (or `tokio`) feature flag.
- First-party integration is Tokio: `tokio::sync::mpsc` for mailboxes, `tokio::sync::oneshot` for replies, `tokio::sync::Semaphore` for the writer slot.
- Worker threads are plain OS threads via `std::thread::spawn`, not Tokio's blocking pool. This guarantees per-connection thread identity (SQLite's multi-thread mode requires that each connection be used from a single thread at a time) and avoids entanglement with Tokio's blocking-thread accounting.
- Alternative runtimes (`futures` channels, `async-std`) can be added later as parallel feature flags.
## Implementation roadmap
Each phase is independently shippable.
1. **`Reader` / `Writer` traits, `Query` / `Mutation` traits.** No behavior change; extract vocabulary from `Connection` into trait defaults. Adopt the GAT-based `Query` shape from the `query` branch; add `Mutation` as a parallel trait. Update `#[derive(Query)]` macro to classify SQL and emit the appropriate trait impl (plus the `OwnedOutput` impl). Tests: existing sync tests continue to pass; add tests for `Mutation` emission.
2. **Sync `Transaction` types.** `ReadTransaction`, `WriteTransaction` with `BEGIN IMMEDIATE` default. `Drop` = rollback. Trait impls wire through to the inner connection. Tests for both commit and drop-rollback paths; test that `read_tx.mutate::<M>(...)` is a compile error (trybuild).
3. **Sync `Pool`.** `PoolBuilder`, writer slot + reader pool, `busy_timeout` default, `PooledReader` / `PooledWriter` guards. WAL mode initialization. Tests: N threads hammering the pool with mixed reads and writes; assert no internal `SQLITE_BUSY`.
4. **Feature gate, async scaffolding.** `async` feature flag. `src/ffi/interrupt.rs`. Shared actor message infrastructure.
5. **`AsyncConnection`.** One actor per connection, Tokio mailbox, drop-interrupt. `AsyncReader` / `AsyncWriter` impls. `run` / `mutate` / `with_reader` / `with_writer` first; `stream` / `mutate_stream` second.
6. **Per-connection statement cache.** `TypeId`-keyed LRU. Tests: instrumented `prepare_v2` counter confirming second `run::<Q>` does not re-prepare.
7. **`AsyncPool`.** N reader actors + 1 writer actor, routing logic. Shares `PoolBuilder` with the sync pool. Tests: many concurrent tasks submitting reads and writes; drop-cancellation test verifying `sqlite3_interrupt` fires and the worker returns to a clean state.
8. **Async transactions.** `AsyncReadTransaction` / `AsyncWriteTransaction` — guards that hold the reader/writer slot. `Drop` dispatches a fire-and-forget rollback.
9. **Docs + examples.** Rustdoc pass on every public type. `examples/axum.rs` demonstrating an HTTP server with drop-cancellation on client disconnect.