From 4aadf33af39befe56ff326301858e9427e9b5d61 Mon Sep 17 00:00:00 2001 From: Kara McCalley Date: Thu, 17 Sep 2026 17:23:00 -0400 Subject: [PATCH] fix: publish changes once the committing statement finishes SQLite's commit hook runs before the commit is final, so a subscriber notified from it can read pre-commit state through another connection. A concurrent reader saw the stale value on 44 of 50 notifications. The commit hook now converts and holds the changes. A trace hook publishes them when the statement that committed finishes, and the commit is durable by then. The trace hook also fires when the next statement starts, which publishes a commit made during a statement reset: an autocommit INSERT ... RETURNING read with fetch_one commits inside the reset, after the profile event. The writer does nothing extra. If the commit fails after the commit hook ran, the rollback hook discards the held changes. Publication now happens after SQLite released the write lock, so notifications keep commit order per connection only. The conn-mgr writer is exclusive, so the plugin keeps its order. Per-transaction state moves from the shared broker to each connection's hook context, so two connections observing the same database can no longer publish or discard each other's changes. The broker's on_preupdate, on_commit, and on_rollback methods are removed, and PreUpdateEvent is crate-private. The CHANGELOG records the breaks. Five tests guard this: visibility in each journal mode, cross-connection isolation, a reset commit surviving a later failed write, and a reset commit published on drop. The first four fail on master. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 10 +- README.md | 3 +- crates/sqlx-sqlite-observer/README.md | 100 ++++-- crates/sqlx-sqlite-observer/src/broker.rs | 112 ++---- crates/sqlx-sqlite-observer/src/conn_mgr.rs | 52 +-- crates/sqlx-sqlite-observer/src/connection.rs | 8 +- crates/sqlx-sqlite-observer/src/hooks.rs | 333 ++++++++++++++---- crates/sqlx-sqlite-observer/src/lib.rs | 8 +- crates/sqlx-sqlite-observer/src/observer.rs | 7 +- .../tests/conn_mgr_tests.rs | 5 +- .../tests/integration_tests.rs | 309 ++++++++++++++++ .../tests/observation_tests.rs | 10 +- 12 files changed, 718 insertions(+), 239 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ba3047..89fef81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,12 +44,15 @@ Databases must be registered on the Rust side with a stable key before they can - **`DatabaseWrapper::close()` and `remove()` now take `self`** instead of `mut self`. Source-compatible for existing callers. - **`ObservableWriteGuard::into_inner()` now returns the new `sqlx_sqlite_observer::UnobservedWriter` enum** (`Regular(WriteGuard)` or `Attached(AttachedWriteGuard)`) instead of unconditionally returning `WriteGuard`. Had no callers anywhere in this workspace prior to this release. - **`sqlx_sqlite_observer::hooks::register_hooks()` now takes a schema-alias → broker `HashMap`** instead of a single `Arc`, to support routing attached-database writes to their owning database's broker. -- **`sqlx_sqlite_observer::TableChange` and `sqlx_sqlite_observer::hooks::PreUpdateEvent` each gain a `pub schema: String` field** (#53) — `"main"` for the primary database, the `ATTACH` alias otherwise. See `TableChange::schema`'s own docs for why it is provenance metadata rather than a stable identifier. **Both types are now also `#[non_exhaustive]`**, so downstream struct literals and exhaustive destructuring patterns need updating (read fields, or add `..` to patterns). Only this crate constructs either type, so in practice this affects code that pattern-matches their full shape. Deliberately unlike `Error` above, which stays exhaustive: these two are expected to gain fields again, and `#[non_exhaustive]` makes the next addition a non-event rather than another major bump. +- **`sqlx_sqlite_observer::TableChange` gains a `pub schema: String` field** (#53) — `"main"` for the primary database, the `ATTACH` alias otherwise. See `TableChange::schema`'s own docs for why it is provenance metadata rather than a stable identifier. **The type is now also `#[non_exhaustive]`**, so downstream struct literals and exhaustive destructuring patterns need updating (read fields, or add `..` to patterns). Only this crate constructs it, so in practice this affects code that pattern-matches its full shape. Deliberately unlike `Error` above, which stays exhaustive: it is expected to gain fields again, and `#[non_exhaustive]` makes the next addition a non-event rather than another major bump. (`hooks::PreUpdateEvent` gained the same field, but is no longer public - see below.) - **`AttachedSpec::schema_name` is now capped at 64 bytes**, failing with `sqlx_sqlite_conn_mgr::Error::InvalidSchemaName` above that (which reaches the frontend as `CONNECTION_ERROR`, or `OBSERVER_ERROR` when observation is active — see the Frontend API note above). Shape was already validated (`[A-Za-z0-9_]`, no leading digit) but length was not, and the alias is now copied into every change notification for the owning database and serialized to each of its subscribers, once per changed row. - **`AttachedSpec::schema_name` values of `main`/`temp` (case-insensitively) are now rejected with `Error::InvalidSchemaName`**, and two specs sharing an alias - compared case-insensitively - now fail with the new `Error::DuplicateSchemaName` instead of a raw SQLite `ATTACH` error (see Fixed below for why this also closes a stranding vector). - **`Error::InvalidSchemaName`'s message now lists every rule it enforces** — non-empty, `[A-Za-z0-9_]` only, no leading digit, at most 64 bytes, and not the reserved `main`/`temp`. It previously named only the character-shape rules, so a rejected `main` (or an over-long alias) produced an error whose every stated condition the input satisfied. The variant is unchanged; only the message text is, which matters for anything matching on the string rather than the variant. - Added **`sqlx_sqlite_conn_mgr::Error::DuplicateSchemaName`** (surfaces to the frontend the same way `InvalidSchemaName` does — `CONNECTION_ERROR`, or `OBSERVER_ERROR` when observation is active). `Error` is not `#[non_exhaustive]`, so exhaustive matches on it need a new arm. - Added **`sqlx_sqlite_observer::Error::BrokerAliasCollision`** (feature-gated on `conn-mgr`). Like `sqlx_sqlite_conn_mgr::Error`, the observer's `Error` is not `#[non_exhaustive]`, so exhaustive matches need a new arm. It guards the alias → broker map against being built with a colliding key; `validate_attached_specs()` now rejects every input that could reach it, so it is defence-in-depth against a future regression in `ATTACH` ordering rather than a reachable error today. +- **`sqlx_sqlite_observer::ObservationBroker::on_preupdate()`, `on_commit()`, and `on_rollback()` are removed.** All three were `pub`. Per-transaction buffering moved from the broker to the per-connection hook context that `register_hooks()` installs, so a caller that drove a broker directly has no replacement: that hook context, not the broker, now owns every piece of per-transaction state. The broker keeps only the state it shares across connections - the observed tables, the schema information, the event conversion, and the broadcast channel. + - **`sqlx_sqlite_observer::hooks::PreUpdateEvent` is now crate-private.** With `on_preupdate()` gone, no public function took or returned it. + - **`sqlx_sqlite_observer::unregister_hooks()` does more.** It still unregisters the preupdate, commit, and rollback hooks and reclaims the hook context. It now also clears the connection's `sqlite3_trace_v2` slot, and publishes any committed change the trace hook has not published yet (a commit made during a statement reset, with no statement run since). A caller who replaced this crate's trace hook with their own loses that hook here. ### Added @@ -109,6 +112,11 @@ Transaction cleanup failures propagate as errors rather than being logged and ig - **Attached-database observation no longer requires the main database to be observed too.** The whole observable acquisition path was gated on the *main* database's broker, so a write through an unobserved database into an observed `readWrite` attachment registered no hooks at all and reached that attachment's own subscribers not at all — silently, with every call reporting success. This was the same class of defect as #53 itself: observation depending on the wrong object. The observable path is now taken when *any* participating database is observed — main or a `readWrite` attachment — and `main` is simply omitted from the alias → broker map when the main database has no broker of its own, so its changes are dropped rather than misattributed. Nothing had covered this configuration, because the attached-observation tests enabled observation on both databases. - **Known limitation:** a writer holds the broker it bound at acquisition for its whole lifetime, so a `disable_observation()` + `enable_observation()` cycle *during* an open transaction leaves that transaction publishing to the previous broker. Subscribers created before the cycle still receive its commit (the hook context keeps that broker alive); subscribers created after the cycle do not, while `is_observing()`, `subscribe()`, and the commit itself all report success. Reachable as: the last window `unobserve()`s and then `observe()`s again while another caller's interruptible transaction is still open. Pinned by an `#[ignore]`d test rather than only prose; the fix (reviving the slot's previous broker, gated on a lease held for the hooks' lifetime) is follow-up work. - **Known limitation:** the plugin's `observe()`/`unobserve()` reference count is keyed on webview labels, so a Rust caller that enabled observation directly on a `DatabaseWrapper` is outside it. The last window releasing tears observation down database-wide and silently ends that caller's subscription; conversely a Rust `disable_observation()` breaks every window's live subscription while leaving their registrations non-zero. A plugin-level API for Rust observers is follow-up work. +- **A change notification now means that the change is durable and readable, not only that the commit hook ran.** SQLite's commit hook runs before the commit is final. It is the hook whose non-zero return converts a `COMMIT` into a `ROLLBACK`. A subscriber that read immediately after notification previously saw pre-commit state through another connection, on most notifications. The tests `a_notification_means_the_change_is_readable_in_wal_mode` and `a_notification_means_the_change_is_readable_in_rollback_journal_mode` now guard this, one per journal mode. The commit hook now converts and holds the transaction's changes and does not publish them yet. A hook registered with `SQLITE_TRACE_PROFILE` and `SQLITE_TRACE_STMT` publishes the changes when the statement that committed finishes, or when the next statement starts. By then the commit is durable, so any connection can read it. The writer does nothing extra. + - **Notification order is now per connection.** Publication happens after SQLite released the write lock, so two connections that commit in one order can notify in the other. On master, the commit hook published under the lock, which kept commit order across connections. Every write through `sqlx-sqlite-conn-mgr`'s single writer, and so every write through the plugin, keeps its order. + - **A commit made during a statement reset notifies when the next statement starts.** `fetch_one` stops at the first row and resets the statement, and for an autocommit `INSERT ... RETURNING` the commit happens inside that reset, after SQLite's profile event. The trace hook is therefore also registered for `SQLITE_TRACE_STMT`, which fires when a statement starts, before any of its hooks. Without it, the held change waited for the next statement's profile event, and a failed next statement's rollback hook discarded it: the row was committed and no notification ever arrived. The test `a_commit_made_during_a_statement_reset_survives_a_later_failed_write` guards this. Dropping the connection also publishes it, guarded by `dropping_a_connection_publishes_a_commit_made_during_a_statement_reset`. + - The trace hook occupies the connection's one `sqlite3_trace_v2` slot. sqlx does not use that slot. If a consumer installs a trace hook on an observed connection, the new hook replaces this crate's hook, and notifications from that connection stop. + - Buffering is now per connection instead of per broker, so an abandoned transaction cannot surface as a phantom change on any other connection's commit. The mid-transaction fix above (#53) covered the writer guards of `sqlx-sqlite-conn-mgr`, which discarded the shared buffer when they came down. A connection from the plain sqlx pool (`SqliteObserver::acquire()`) had no such discard, and one shared buffer per database meant its stranded events surfaced on whichever connection committed next. The buffer now dies with the hook context on both paths. - **`ObservableSqliteDatabase::acquire_writer()` no longer holds the single write permit while waiting on a read-pool connection.** It warmed each observed table's cached schema info *after* taking the permit, and that warm-up awaits the read pool (six connections by default). Six tasks each holding a read connection and then waiting for the writer would deadlock against it until sqlx's 30-second acquire timeout fired, failing up to seven of them with `PoolTimedOut`. Reachable from an ordinary query-then-write pattern, and on every acquisition — not just the first — whenever an observed table doesn't exist in the schema, since that result is never cached. The attached-writer path already warmed first; both now do. **The trade this makes:** a table that joins the observed set after warming — via `subscribe()`, `subscribe_stream()`, or a later `observe()`, including *mid-transaction*, since the preupdate hook checks the observed set live when a change fires rather than snapshotting it at hook registration — has its changes published with an empty `primary_key`, and a meaningless `rowid` if the table is `WITHOUT ROWID`, converging on the next writer acquisition. Re-checking after the permit is acquired would not close that window (it ends at end-of-transaction, not at permit acquisition) and would reintroduce the deadlock, because an observed name that doesn't resolve in the schema is never cached and so would take the read pool under the permit on every acquisition. - **An attached-database query whose statement fails now still detaches.** `execute`, `fetch_all`, `fetch_one`, `fetch_page` and both transaction builders returned early on a statement error, or on a failed `BEGIN IMMEDIATE`, without issuing `DETACH`. (A failing `COMMIT` or `ROLLBACK` still strands the alias — a narrower case, since releasing it there means rolling back first, and the error to report becomes a judgement call. Tracked separately. This was not, however, the only remaining stranding vector - acquiring the attached connection or writer in the first place could strand an alias before a single query ever ran, see the next entry.) Neither the attached guards' `Drop` nor the pool's `after_release` hook detaches — `after_release` runs `ROLLBACK` only — so the alias stayed bound to the pooled connection. Since the write pool holds a single connection, every later attach of that alias then failed with "database ... is already in use" until the connection went genuinely idle (30 seconds by default), which a retry loop prevents indefinitely. - **`acquire_reader_with_attached()` and `acquire_writer_with_attached()` no longer strand an alias when a later spec in the same call is rejected or fails to `ATTACH`.** Three related gaps in the same invariant as the entry above: diff --git a/README.md b/README.md index 8894d4a..f050cb6 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,8 @@ without Tauri: * **[`sqlx-sqlite-conn-mgr`](crates/sqlx-sqlite-conn-mgr/)** — Low-level connection management: read pool, exclusive writer, WAL mode, attached databases * **[`sqlx-sqlite-observer`](crates/sqlx-sqlite-observer/)** — Reactive change - notifications using SQLite's native preupdate/commit/rollback hooks + notifications using SQLite's native `preupdate_hook`, `commit_hook`, + `rollback_hook`, and a `SQLITE_TRACE_PROFILE` trace hook * **[`sqlx-sqlite-toolkit`](crates/sqlx-sqlite-toolkit/)** — High-level API: `DatabaseWrapper`, builder-pattern queries, interruptible transactions, JSON type decoding. Optionally integrates the observer behind a feature flag. diff --git a/crates/sqlx-sqlite-observer/README.md b/crates/sqlx-sqlite-observer/README.md index a367047..0878cc0 100644 --- a/crates/sqlx-sqlite-observer/README.md +++ b/crates/sqlx-sqlite-observer/README.md @@ -3,11 +3,12 @@ Reactive change notifications for SQLite databases using sqlx. This crate provides **transaction-safe** change notifications for SQLite databases -using SQLite's native hooks (`preupdate_hook`, `commit_hook`, `rollback_hook`). +using SQLite's native hooks (`preupdate_hook`, `commit_hook`, `rollback_hook`, and a +`SQLITE_TRACE_PROFILE` trace hook). ## Features - * **Transaction-safe notifications**: Changes only notify after successful commit + * **Transaction-safe notifications**: Changes notify only once committed and readable * **Typed column values**: Access old/new values with native SQLite types * **Stream support**: Use `tokio_stream::Stream` for async iteration * **Multiple subscribers**: Broadcast channel supports multiple listeners @@ -57,38 +58,43 @@ sqlx-sqlite-observer = { version = "0.8", features = ["conn-mgr"] } The library uses SQLite's native hooks for transaction-safe change tracking: ```text -┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ -│ preupdate_hook │────►│ broker.buffer │ │ subscribers │ -│ (captures data) │ │ (Vec) │ │ │ -└─────────────────┘ └────────┬────────┘ └─────────────────┘ - │ ▲ - ┌────────────| │ - │ │ │ - ┌─────▼────┐ ┌────▼─────┐ │ - │ COMMIT │ │ ROLLBACK │ │ - └─────┬────┘ └────┬─────┘ │ - │ │ │ - ▼ ▼ │ - on_commit() on_rollback() │ - │ │ │ - │ buffer.clear() │ - │ (discard) │ - │ │ - └────────────────────────────────────┘ +┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ +│ preupdate_hook │────►│ context.buffer │ │ subscribers │ +│ (captures data) │ │ (per connection) │ │ │ +└─────────────────┘ └────────┬─────────┘ └─────────────────┘ + │ ▲ + ┌────────────┼────────────┐ │ + │ │ │ │ + ┌─────▼────┐ ┌────▼─────┐ │ │ + │ COMMIT │ │ ROLLBACK │ │ │ + └─────┬────┘ └────┬─────┘ │ │ + │ │ │ │ + ▼ ▼ │ │ + commit_hook rollback_hook │ │ + convert + hold buffer.clear() │ │ + │ (discard) │ │ + ▼ │ │ + trace_hook (statement finished) │ │ + broker.publish() ──────────────────┴───────────┘ change_tx.send() - (publish) ``` 1. When you acquire a connection, observation hooks are registered on the raw - SQLite handle + SQLite handle. Each connection gets its own buffer, so two connections that + observe the same database never publish or discard each other's changes. 2. `preupdate_hook` captures changes (table, operation, old/new values) and - buffers them -3. `commit_hook` fires when a transaction commits, publishing buffered changes - to subscribers -4. `rollback_hook` fires when a transaction rolls back, discarding buffered - changes - -This ensures subscribers **only receive notifications for committed changes**. + buffers them. +3. `commit_hook` fires when a transaction is about to commit. It converts the + buffered changes and holds them. It does not publish, because it runs before + the commit is final. +4. The trace hook fires when the statement that committed finishes. The commit + is durable by then, so the held changes are published. +5. `rollback_hook` fires when a transaction rolls back, discarding buffered + changes. + +This ensures subscribers **only receive notifications for committed changes**, +and that a subscriber who reads as soon as it is notified reads the committed +state. ## API Reference @@ -320,7 +326,8 @@ async fn main() -> Result<(), Box> { .await?; sqlx::query("COMMIT").execute(&mut *writer).await?; - // Notification arrives after commit + // The notification arrives once the COMMIT has finished, so the change is + // readable through any connection. let change = rx.recv().await?; println!("Changed: {}", change.table); @@ -328,6 +335,39 @@ async fn main() -> Result<(), Box> { } ``` +## When Notifications Arrive + +A notification means the change is committed and readable through any +connection. Nothing is required of the writer for that to hold. It takes two +hooks, because SQLite's commit hook runs *before* the commit is final. It is +the hook whose non-zero return converts a `COMMIT` into a `ROLLBACK`. + +1. The commit hook converts the buffered changes and holds them. +2. A trace hook, registered with `SQLITE_TRACE_PROFILE` and + `SQLITE_TRACE_STMT`, fires when the statement that committed finishes + running, and again when the next statement starts. The commit is durable by + then, and the held changes are published. + +If publication happened in the commit hook instead, a subscriber that reads as +soon as it is notified would read pre-commit state. + +Two consequences follow from publishing after the commit: + + * **Order is per connection.** Publication happens after SQLite released the + write lock, so two connections that commit in one order can notify in the + other. Writes through one connection, including every write through + `sqlx-sqlite-conn-mgr`'s single writer, notify in commit order. + * **A commit made during a statement reset notifies late.** `fetch_one` stops + at the first row and resets the statement. For an autocommit + `INSERT ... RETURNING`, the commit happens inside that reset, after the + profile event. The trace hook is also registered for `SQLITE_TRACE_STMT`, + so the change is published when the next statement on that connection + starts, or when the connection is released, whichever comes first. + +The trace hook occupies the connection's single `sqlite3_trace_v2` slot. sqlx +does not use that slot. If you install your own trace hook on an observed +connection, notifications from that connection stop. + ## Usage Notes ### Channel Capacity diff --git a/crates/sqlx-sqlite-observer/src/broker.rs b/crates/sqlx-sqlite-observer/src/broker.rs index fdcf696..55213b6 100644 --- a/crates/sqlx-sqlite-observer/src/broker.rs +++ b/crates/sqlx-sqlite-observer/src/broker.rs @@ -1,14 +1,20 @@ -//! Transaction-aware observation broker for buffering and publishing changes. +//! Transaction-aware observation broker for converting and publishing changes. //! //! This module provides transaction-safe change notifications. Changes are buffered //! during transactions (explicit and implicit) and only published after successful -//! commit. Rolled-back transactions produce no notifications. +//! commit, once the committing statement finishes. Rolled-back transactions produce +//! no notifications. +//! +//! The buffer lives in each connection's hook context (see [`crate::hooks`]), not +//! on the broker, so two connections observing the same database never publish or +//! discard each other's changes. The broker holds what every connection shares: +//! observed tables, schema info, event conversion, and the broadcast channel. //! //! # Data Flow //! //! ```text //! ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ -//! │ preupdate_hook │────►│ broker.buffer │ │ subscribers │ +//! │ preupdate_hook │────►│ context.buffer │ │ subscribers │ //! │ (captures data) │ │ (Vec) │ │ │ //! └─────────────────┘ └────────┬────────┘ └─────────────────┘ //! │ ▲ @@ -19,10 +25,12 @@ //! └─────┬────┘ └────┬─────┘ │ │ //! │ │ │ │ //! ▼ ▼ │ │ -//! on_commit() on_rollback() │ │ -//! │ │ │ │ +//! commit_hook rollback_hook │ │ +//! (convert, hold) │ │ │ //! │ buffer.clear() │ │ -//! │ (discard) │ │ +//! ▼ (discard) │ │ +//! trace_hook │ │ +//! (statement finished) │ │ //! │ │ │ //! └─────────────────────────┴──────────┘ //! change_tx.send() @@ -30,27 +38,31 @@ //! ``` //! //! Changes captured by the preupdate hook are buffered until the transaction -//! (explicit or implicit) completes. On commit, buffered changes are published -//! to subscribers. On rollback, they are discarded without notification. +//! (explicit or implicit) completes. On commit, the commit hook converts the +//! buffered changes and holds them, and the trace hook publishes them to +//! subscribers when the committing statement finishes. On rollback, they are +//! discarded without notification. use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::time::Instant; -use parking_lot::{Mutex, RwLock}; +use parking_lot::RwLock; use tokio::sync::broadcast; -use tracing::{debug, error, trace}; +use tracing::trace; use crate::change::{ChangeOperation, ColumnValue, TableChange, TableInfo}; use crate::hooks::{PreUpdateEvent, SqliteValue}; -/// Transaction-aware observation broker. +/// Shared observation broker for one database. /// -/// Buffers preupdate events during transactions and publishes them to -/// subscribers only after successful commit. Rolled-back transactions -/// have their buffered changes discarded. +/// Knows which tables are observed and their schema, converts a connection's +/// captured events into [`TableChange`]s, and publishes them to subscribers. +/// A broker is shared by every connection that observes its database; the +/// per-transaction buffering lives in each connection's hook context (see the +/// module doc), so a rolled-back or abandoned transaction on one connection +/// never touches another's changes. pub struct ObservationBroker { - buffer: Mutex>, change_tx: broadcast::Sender, observed_tables: RwLock>, table_info: RwLock>, @@ -72,7 +84,6 @@ impl ObservationBroker { assert!(channel_capacity > 0, "channel_capacity must be at least 1"); let (change_tx, _) = broadcast::channel(channel_capacity); Arc::new(Self { - buffer: Mutex::new(Vec::new()), change_tx, observed_tables: RwLock::new(HashSet::new()), table_info: RwLock::new(HashMap::new()), @@ -182,73 +193,29 @@ impl ObservationBroker { self.observed_tables.read().iter().cloned().collect() } - /// Called by preupdate_hook - buffers the event for later processing. + /// Publishes one converted change to every subscriber. /// - /// Events are held in the buffer until either `on_commit()` (publish) - /// or `on_rollback()` (discard) is called. - pub fn on_preupdate(&self, event: PreUpdateEvent) { - trace!( - table = %event.table, - operation = ?event.operation, - "Buffering preupdate event" - ); - self.buffer.lock().push(event); - } - - /// Called by commit_hook - flushes buffered events to subscribers. - /// - /// Converts all buffered `PreUpdateEvent`s to `TableChange`s and sends - /// them through the broadcast channel. The buffer is cleared afterward. - pub fn on_commit(&self) { - let events: Vec = { - let mut buffer = self.buffer.lock(); - std::mem::take(&mut *buffer) - }; - - if events.is_empty() { - return; - } - - debug!(count = events.len(), "Flushing buffered changes on commit"); - - for event in events { - match self.event_to_change(event) { - Ok(table_change) => { - let _ = self.change_tx.send(table_change); - } - Err(e) => { - error!(error = %e, "Failed to convert event to change"); - } - } - } - } - - /// Called by rollback_hook - discards all buffered events. - /// - /// Clears the buffer without publishing any changes to subscribers. - pub fn on_rollback(&self) { - let count = { - let mut buffer = self.buffer.lock(); - let count = buffer.len(); - buffer.clear(); - count - }; - - if count > 0 { - debug!(count, "Discarding buffered changes on rollback"); - } + /// Called by a connection's trace hook once the statement that committed the + /// change has finished, so a subscriber that reads on notification reads the + /// committed state. A send error means there are no subscribers, which is not + /// an error for the writer. + pub(crate) fn publish(&self, change: TableChange) { + let _ = self.change_tx.send(change); } /// Subscribes to change notifications. /// /// Returns a broadcast receiver that will receive `TableChange` events - /// after transactions commit. + /// once a committed transaction's changes are readable. pub fn subscribe(&self) -> broadcast::Receiver { self.change_tx.subscribe() } /// Converts a PreUpdateEvent to a TableChange for broadcast. - fn event_to_change(&self, event: PreUpdateEvent) -> crate::Result { + /// + /// Called by a connection's commit hook, which needs this broker's schema + /// information for the table the event touched. + pub(crate) fn event_to_change(&self, event: PreUpdateEvent) -> crate::Result { let table_info = self.table_info.read().get(&event.table).cloned(); // For WITHOUT ROWID tables, the rowid from preupdate hook is not meaningful @@ -338,7 +305,6 @@ impl ObservationBroker { impl std::fmt::Debug for ObservationBroker { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ObservationBroker") - .field("buffer_len", &self.buffer.lock().len()) .field("observed_tables", &self.observed_tables.read().len()) .field("channel_capacity", &self.channel_capacity) .field("capture_values", &self.capture_values) diff --git a/crates/sqlx-sqlite-observer/src/conn_mgr.rs b/crates/sqlx-sqlite-observer/src/conn_mgr.rs index 9f385df..19d2cfa 100644 --- a/crates/sqlx-sqlite-observer/src/conn_mgr.rs +++ b/crates/sqlx-sqlite-observer/src/conn_mgr.rs @@ -4,7 +4,8 @@ //! `sqlx-sqlite-conn-mgr`. Enable with the `conn-mgr` feature. //! //! Uses SQLite's native hooks for transaction-safe change tracking. Changes -//! are buffered during transactions and only published after commit. +//! are buffered during transactions and published once the statement that +//! committed them finishes, when they are readable. //! //! # Example //! @@ -226,7 +227,6 @@ impl ObservableSqliteDatabase { writer: Some(InnerWriter::Regular(writer)), hooks_registered: false, raw_db: None, - brokers: HashMap::new(), }; let mut brokers = HashMap::with_capacity(1); @@ -469,7 +469,6 @@ pub async fn acquire_writer_with_attached_brokers( writer: Some(InnerWriter::Attached(writer)), hooks_registered: false, raw_db: None, - brokers: HashMap::new(), }; // `brokers` can be empty here even though a caller's own gate found at least @@ -576,13 +575,6 @@ pub struct ObservableWriteGuard { /// call unregister_hooks synchronously in Drop without needing /// the async lock_handle. raw_db: Option<*mut sqlite3>, - /// Brokers hooks were registered with, keyed by schema alias. Retained - /// (rather than discarded once `hooks::register_hooks` has its own copy) - /// so `Drop` can discard each one's buffered-but-uncommitted events if - /// this guard is dropped without an explicit commit or rollback - see - /// `Drop`'s impl for why that's safe to do unconditionally. Empty until - /// `register_hooks` populates it. - brokers: HashMap>, } // SAFETY: The raw_db pointer is only used for hook registration/unregistration @@ -613,7 +605,7 @@ impl ObservableWriteGuard { let db: *mut sqlite3 = handle.as_raw_handle().as_ptr(); unsafe { - hooks::register_hooks(db, brokers.clone())?; + hooks::register_hooks(db, brokers)?; } // Cache the raw pointer so Drop can call unregister_hooks synchronously. @@ -621,37 +613,17 @@ impl ObservableWriteGuard { // which we own via self.writer. self.raw_db = Some(db); self.hooks_registered = true; - self.brokers = brokers; Ok(()) } - /// Discards every broker's buffered-but-uncommitted events. - /// - /// Safe to call unconditionally, regardless of whether a commit or - /// rollback already ran: `on_commit` drains the buffer via `mem::take` - /// before publishing, and an explicit `ROLLBACK`'s own rollback_hook - /// already clears it too, so calling this afterward always finds nothing - /// left to discard. The only case where it does something is the one it - /// exists for: hooks torn down - by [`Drop`](Self), [`into_inner`], or - /// [`detach_all`] - with no commit or rollback ever having run, which - /// would otherwise let this transaction's buffered events resurface as - /// phantom changes on the *next* transaction's commit. - /// - /// [`into_inner`]: Self::into_inner - /// [`detach_all`]: Self::detach_all - fn flush_all_brokers(&self) { - for broker in self.brokers.values() { - broker.on_rollback(); - } - } - /// Consumes this wrapper and returns the underlying write guard. /// /// Hooks are unregistered before returning the guard, so it can be - /// safely used without observation. Also flushes every broker's buffer - /// (see [`flush_all_brokers`](Self::flush_all_brokers)) - safe to call - /// whether or not a commit/rollback already ran, and necessary if this is - /// called mid-transaction, with no commit or rollback yet sent. + /// safely used without observation. Unregistering also drops this writer's + /// buffered-but-uncommitted events (see [`hooks::unregister_hooks`]), which + /// matters if this is called mid-transaction, with no commit or rollback yet + /// sent: those events would otherwise resurface as phantom changes on the + /// connection's next commit. pub fn into_inner(mut self) -> UnobservedWriter { // Unregister hooks before returning the writer to prevent // use-after-free if the broker is dropped before the connection is reused. @@ -662,7 +634,6 @@ impl ObservableWriteGuard { crate::hooks::unregister_hooks(db); } trace!("Hooks unregistered before returning inner writer"); - self.flush_all_brokers(); } self.hooks_registered = false; self.raw_db = None; @@ -676,9 +647,8 @@ impl ObservableWriteGuard { /// /// If this guard wraps a plain (non-attached) writer, there is nothing to /// detach - this reduces to hook unregistration, safe to call regardless - /// of which kind of writer this guard wraps. Also flushes every broker's - /// buffer (see [`flush_all_brokers`](Self::flush_all_brokers)) - safe to - /// call whether or not a commit/rollback already ran. + /// of which kind of writer this guard wraps. Unregistering also drops this + /// writer's buffered-but-uncommitted events (see [`hooks::unregister_hooks`]). pub async fn detach_all(mut self) -> Result<()> { if self.hooks_registered && let Some(db) = self.raw_db @@ -687,7 +657,6 @@ impl ObservableWriteGuard { crate::hooks::unregister_hooks(db); } trace!("Hooks unregistered before detach_all"); - self.flush_all_brokers(); } self.hooks_registered = false; self.raw_db = None; @@ -712,7 +681,6 @@ impl Drop for ObservableWriteGuard { hooks::unregister_hooks(db); } trace!("ObservableWriteGuard dropped, hooks unregistered"); - self.flush_all_brokers(); } } } diff --git a/crates/sqlx-sqlite-observer/src/connection.rs b/crates/sqlx-sqlite-observer/src/connection.rs index 5fbe085..79d120b 100644 --- a/crates/sqlx-sqlite-observer/src/connection.rs +++ b/crates/sqlx-sqlite-observer/src/connection.rs @@ -1,8 +1,8 @@ //! Observable connection wrapper with SQLite hook integration. //! //! Provides change tracking via SQLite's native preupdate/commit/rollback hooks -//! instead of triggers. Changes are buffered during transactions and only -//! published to subscribers after successful commit. +//! instead of triggers. Changes are buffered during transactions and published +//! once the statement that committed them finishes, when they are readable. use std::ops::{Deref, DerefMut}; use std::sync::Arc; @@ -19,10 +19,6 @@ use crate::hooks; /// A wrapper around a SQLite pool connection allowing observers to subscribe to /// change notifications. /// -/// Uses SQLite's native hooks (preupdate_hook, commit_hook, rollback_hook) -/// for transaction-safe change tracking. Changes are buffered during transactions -/// and published to subscribers only after successful commit. -/// /// Implements `Deref`/`DerefMut` to allow transparent use as the underlying /// `PoolConnection`. pub struct ObservableConnection { diff --git a/crates/sqlx-sqlite-observer/src/hooks.rs b/crates/sqlx-sqlite-observer/src/hooks.rs index bb41cfe..3ae3953 100644 --- a/crates/sqlx-sqlite-observer/src/hooks.rs +++ b/crates/sqlx-sqlite-observer/src/hooks.rs @@ -1,7 +1,27 @@ //! SQLite native hook registration for support observing changes to the database. //! //! This module provides low-level bindings to SQLite's preupdate_hook, commit_hook, -//! and rollback_hook APIs for transaction-aware change tracking. +//! rollback_hook, and trace APIs for transaction-aware change tracking. +//! +//! # Why four hooks +//! +//! SQLite's commit hook runs *before* the commit is final: it is the hook whose +//! non-zero return converts a `COMMIT` into a `ROLLBACK`. A subscriber notified +//! from it and reading through another connection reads pre-commit state. So +//! the commit hook only converts and holds the transaction's changes, and the +//! trace hook publishes them. It listens for two events: a statement finishing, +//! which is after any commit it made became durable, and a statement starting, +//! which catches a commit that happened during a statement reset (an autocommit +//! `INSERT ... RETURNING` read with `fetch_one`). A reset commit is published +//! when the next statement starts or when the hooks are unregistered. +//! +//! Publication happens after SQLite released the write lock, so notifications +//! keep commit order per connection only. +//! +//! All per-transaction state lives in the `HookContext` that each connection +//! registers, never on the shared broker, so two connections observing the +//! same database can never publish or discard each other's changes. The data +//! flow diagram in [`crate::broker`] shows the whole path. //! //! # SQLite Requirements //! @@ -10,23 +30,25 @@ //! SQLite library supports this feature. use std::collections::HashMap; -use std::ffi::{CStr, CString, c_char, c_int, c_void}; +use std::ffi::{CStr, CString, c_char, c_int, c_uint, c_void}; use std::io::Write; use std::panic::{AssertUnwindSafe, catch_unwind}; use std::ptr; use std::sync::Arc; use libsqlite3_sys::{ - SQLITE_BLOB, SQLITE_DELETE, SQLITE_FLOAT, SQLITE_INSERT, SQLITE_INTEGER, SQLITE_NULL, - SQLITE_TEXT, SQLITE_UPDATE, sqlite3, sqlite3_commit_hook, sqlite3_compileoption_used, - sqlite3_preupdate_count, sqlite3_preupdate_hook, sqlite3_preupdate_new, sqlite3_preupdate_old, - sqlite3_rollback_hook, sqlite3_value, sqlite3_value_blob, sqlite3_value_bytes, - sqlite3_value_double, sqlite3_value_int64, sqlite3_value_text, sqlite3_value_type, + SQLITE_BLOB, SQLITE_DELETE, SQLITE_FLOAT, SQLITE_INSERT, SQLITE_INTEGER, SQLITE_NULL, SQLITE_OK, + SQLITE_TEXT, SQLITE_TRACE_PROFILE, SQLITE_TRACE_STMT, SQLITE_UPDATE, sqlite3, + sqlite3_commit_hook, sqlite3_compileoption_used, sqlite3_preupdate_count, + sqlite3_preupdate_hook, sqlite3_preupdate_new, sqlite3_preupdate_old, sqlite3_rollback_hook, + sqlite3_trace_v2, sqlite3_value, sqlite3_value_blob, sqlite3_value_bytes, sqlite3_value_double, + sqlite3_value_int64, sqlite3_value_text, sqlite3_value_type, }; +use parking_lot::Mutex; use tracing::{debug, error, trace}; use crate::broker::ObservationBroker; -use crate::change::ChangeOperation; +use crate::change::{ChangeOperation, TableChange}; /// A SQLite value extracted from preupdate hooks. /// @@ -85,12 +107,10 @@ impl SqliteValue { /// Raw change event captured by the preupdate hook before commit decision. /// -/// `#[non_exhaustive]` for the same reason as -/// [`TableChange`](crate::change::TableChange); only `preupdate_callback` builds -/// one. +/// Crate-private: only `preupdate_callback` builds one, and only the broker's +/// conversion consumes one. #[derive(Debug, Clone)] -#[non_exhaustive] -pub struct PreUpdateEvent { +pub(crate) struct PreUpdateEvent { /// The schema this change occurred under (`"main"` or an `ATTACH` alias). /// See [`TableChange::schema`](crate::change::TableChange::schema) for what /// this is - and is not - safe to rely on. @@ -103,16 +123,115 @@ pub struct PreUpdateEvent { pub new_values: Option>, } -/// Context data passed to SQLite hook callbacks. +/// Context data passed to SQLite hook callbacks - one per registered connection. +/// +/// Stored as user_data pointer in SQLite hooks. Brokers are keyed by schema +/// alias (`"main"` for the primary database, the `ATTACH` alias otherwise) +/// rather than held singly, so that one connection with attached databases can +/// route each change to the broker of whichever database actually owns the +/// affected table. The `Arc`s ensure each broker stays alive as long as hooks +/// are registered. /// -/// Stored as user_data pointer in SQLite hooks. Keyed by schema alias -/// (`"main"` for the primary database, the `ATTACH` alias otherwise) rather -/// than holding a single broker, so that one connection with attached -/// databases can route each change to the broker of whichever database -/// actually owns the affected table. The `Arc`s ensure each broker stays -/// alive as long as hooks are registered. +/// The context also owns this connection's transaction state: the events +/// buffered since the last commit or rollback, and the changes a commit hook +/// converted but has not yet published. struct HookContext { brokers: HashMap>, + /// Preupdate events of the transaction in progress. + buffer: Mutex>, + /// Changes the commit hook accepted, awaiting the statement's end to publish, + /// each paired with the broker that converted it. + committed: Mutex, TableChange)>>, +} + +impl HookContext { + fn new(brokers: HashMap>) -> Self { + Self { + brokers, + buffer: Mutex::new(Vec::new()), + committed: Mutex::new(Vec::new()), + } + } + + /// Buffers a captured event until the transaction commits or rolls back. + fn on_preupdate(&self, event: PreUpdateEvent) { + trace!(table = %event.table, operation = ?event.operation, "Buffering preupdate event"); + self.buffer.lock().push(event); + } + + /// Converts the buffered events and holds them for [`publish_committed`]. + /// + /// The buffer is drained before conversion starts, so a panic while + /// converting one event loses that transaction's remaining events but can + /// never let them resurface on the connection's next commit. Lost + /// notifications are the accepted trade for a single panic guard around the + /// whole call, instead of one per event. + /// + /// [`publish_committed`]: Self::publish_committed + fn on_commit(&self) { + let events = std::mem::take(&mut *self.buffer.lock()); + + if events.is_empty() { + return; + } + + debug!(count = events.len(), "Holding buffered changes on commit"); + + let mut committed = self.committed.lock(); + + for event in events { + // The preupdate callback only buffers an event whose schema has a + // broker, so a miss here cannot happen; it is reported rather than + // unwrapped because this runs inside an FFI callback. + let Some(broker) = self.brokers.get(&event.schema) else { + error!(schema = %event.schema, "No broker for a buffered event's schema"); + continue; + }; + + match broker.event_to_change(event) { + Ok(change) => committed.push((Arc::clone(broker), change)), + Err(e) => error!(error = %e, "Failed to convert event to change"), + } + } + } + + /// Discards the buffered events and any held-but-unpublished changes of a + /// rolled-back transaction. + /// + /// A commit can still fail after its commit hook ran, and this hook then fires + /// while that transaction's changes are held. They must not survive it. + fn on_rollback(&self) { + let discarded_buffered = std::mem::take(&mut *self.buffer.lock()).len(); + let discarded_committed = std::mem::take(&mut *self.committed.lock()).len(); + + if discarded_buffered > 0 || discarded_committed > 0 { + debug!( + discarded_buffered, + discarded_committed, "Discarding buffered and held changes on rollback" + ); + } + } + + /// Publishes every held change through its schema's broker. + /// + /// Called by the trace hook when a statement finishes, which is after any + /// commit that statement performed became durable. The trace hook also + /// calls it when the next statement starts, which covers a commit made + /// during a statement reset. [`unregister_hooks`] also calls it as a + /// backstop. Safe to call when nothing is held. + fn publish_committed(&self) { + let changes = std::mem::take(&mut *self.committed.lock()); + + if changes.is_empty() { + return; + } + + debug!(count = changes.len(), "Publishing committed changes"); + + for (broker, change) in changes { + broker.publish(change); + } + } } /// Checks if the linked SQLite library was compiled with `SQLITE_ENABLE_PREUPDATE_HOOK`. @@ -153,7 +272,7 @@ pub fn is_preupdate_hook_enabled() -> bool { /// database; any other key routes changes made under that `ATTACH` alias to /// the corresponding broker. A schema with no entry here has its changes /// silently dropped by the preupdate callback rather than misrouted to some -/// other schema's broker - see [`preupdate_callback`]. +/// other schema's broker - see `preupdate_callback`. /// /// # Errors /// @@ -177,10 +296,10 @@ pub unsafe fn register_hooks( // Heap-allocate the context so it outlives this function. SQLite's C API // requires a raw pointer to pass user data to callbacks. - let context = Box::new(HookContext { brokers }); + let context = Box::new(HookContext::new(brokers)); // Transfer ownership out of Rust's memory management. // - // NOTE: This pointer is shared across all three hooks and is intentionally + // NOTE: This pointer is shared across all four hooks and is intentionally // leaked. SQLite does NOT free user_data - it simply passes the pointer back // to callbacks. The memory is reclaimed when hooks are replaced via // `unregister_hooks`, which reconstructs the Box from the raw pointer returned @@ -192,10 +311,35 @@ pub unsafe fn register_hooks( // SAFETY: db is a valid sqlite3 pointer (guaranteed by caller). // Each hook receives the same context_ptr, which remains valid until // unregister_hooks is called or the process exits. - unsafe { + // + // Two trace events fire (see the module doc). The profile event fires when + // a statement finishes. The statement event fires when the next statement + // starts, and this publishes a commit that happened during a reset. The + // trace slot is single-occupancy. A caller who installs their own trace + // hook on this connection stops notifications. + // + // Unlike the other three, `sqlite3_trace_v2` reports a result. It is the publish + // point, so a failure leaves the connection unobserved rather than half-observed. + let trace_rc = unsafe { sqlite3_preupdate_hook(db, Some(preupdate_callback), context_ptr); sqlite3_commit_hook(db, Some(commit_callback), context_ptr); sqlite3_rollback_hook(db, Some(rollback_callback), context_ptr); + sqlite3_trace_v2( + db, + SQLITE_TRACE_STMT | SQLITE_TRACE_PROFILE, + Some(trace_callback), + context_ptr, + ) + }; + + if trace_rc != SQLITE_OK { + // SAFETY: the same db the hooks were just installed on; nothing has run + // in between, so no callback is in flight. Reclaims the context Box. + unsafe { unregister_hooks(db) }; + return Err(crate::Error::HookRegistration(format!( + "sqlite3_trace_v2 failed with code {trace_rc}; the trace hook publishes \ + notifications, so observation cannot be enabled on this connection" + ))); } trace!("SQLite hooks registered successfully"); @@ -204,6 +348,13 @@ pub unsafe fn register_hooks( /// Unregisters all observation hooks and reclaims the context memory. /// +/// Publishes any change the commit hook accepted that the trace hook has not +/// published yet. This is a commit made during a statement reset, with no +/// statement run since. It also drops whatever was buffered but never +/// committed. The buffer dies with the context, so an abandoned transaction's +/// events cannot resurface on the connection's next commit. This function +/// also clears the connection's trace slot. +/// /// # Safety /// /// - `db` must be the same valid sqlite3 pointer passed to `register_hooks` @@ -217,12 +368,14 @@ pub unsafe fn unregister_hooks(db: *mut sqlite3) { unsafe { sqlite3_commit_hook(db, None, ptr::null_mut()); sqlite3_rollback_hook(db, None, ptr::null_mut()); + sqlite3_trace_v2(db, 0, None, ptr::null_mut()); } // Reclaim the HookContext we leaked in register_hooks if !prev_user_data.is_null() { // SAFETY: prev_user_data was created by Box::into_raw in register_hooks - let _ = unsafe { Box::from_raw(prev_user_data as *mut HookContext) }; + let context = unsafe { Box::from_raw(prev_user_data as *mut HookContext) }; + context.publish_committed(); trace!("SQLite hooks unregistered and context freed"); } } @@ -230,9 +383,9 @@ pub unsafe fn unregister_hooks(db: *mut sqlite3) { /// Preupdate hook callback - captures changes before they're committed. /// /// Called by SQLite for INSERT, UPDATE, and DELETE operations. Captures old/new -/// row values and buffers them in the broker of whichever database owns the -/// affected table (selected by schema - see [`register_hooks`]) until commit -/// or rollback. +/// row values, if the broker of whichever database owns the affected table +/// (selected by schema - see [`register_hooks`]) observes it, and buffers them +/// in the connection's context until commit or rollback. /// /// Note: `user_data` is SQLite's C API term for callback context (our HookContext), /// unrelated to our app's user data. @@ -359,7 +512,7 @@ unsafe extern "C" fn preupdate_callback( new_values, }; - broker.on_preupdate(event); + context.on_preupdate(event); }); if result.is_err() { @@ -374,16 +527,17 @@ unsafe extern "C" fn preupdate_callback( } } -/// Commit hook callback - flushes buffered changes to subscribers. +/// Commit hook callback - converts and holds the transaction's changes. /// /// Called by SQLite when a transaction is about to commit. Returning 0 allows -/// the commit to proceed; returning non-zero would cause a rollback. +/// the commit to proceed; returning non-zero would cause a rollback. Because +/// the commit is not final yet, nothing is published here - see the module +/// doc and [`trace_callback`]. /// /// `sqlite3_commit_hook`'s callback takes no schema argument and fires exactly /// once per transaction regardless of how many schemas (main plus any attached -/// databases) were touched, so every broker in the map must be flushed here - -/// a single-broker flush would strand an attached database's buffered changes -/// with no commit of its own to release them on. +/// databases) were touched. The context buffers every schema's events together +/// and routes each to its broker on conversion, so one call settles them all. /// /// Note: `user_data` is SQLite's C API term for callback context (our HookContext), /// unrelated to application-level user data. @@ -394,36 +548,71 @@ unsafe extern "C" fn commit_callback(user_data: *mut c_void) -> c_int { // SAFETY: user_data is a valid HookContext pointer created in register_hooks. let context = unsafe { &*(user_data as *const HookContext) }; - // Wrapped because it's the one other thing in this fn that can panic, and an - // unwind out of an extern "C" fn aborts the process. Nothing to report if it - // does fail - the panic would be coming from the logger itself. - let _ = catch_unwind(|| trace!("Commit hook fired - flushing changes")); - - // catch_unwind wraps each broker's flush individually rather than the - // whole loop: a panic partway through must not skip flushing the brokers - // that come after it. With a single broker this distinction was moot: the - // fan-out is what makes it matter, since brokers left unflushed here have - // their buffered events resurface as phantom changes on the connection's - // *next* commit rather than this one. - for broker in context.brokers.values() { - if catch_unwind(AssertUnwindSafe(|| broker.on_commit())).is_err() { - // Reported via writeln! to stderr rather than tracing - see - // preupdate_callback's equivalent for why. - let _ = writeln!( - std::io::stderr(), - "sqlx-sqlite-observer: panic in commit_callback (absorbed to keep the process alive)" - ); - } + + // Wrapped because an unwind out of an extern "C" fn aborts the process. + if catch_unwind(AssertUnwindSafe(|| { + trace!("Commit hook fired - holding changes"); + context.on_commit(); + })) + .is_err() + { + // Reported via writeln! to stderr rather than tracing - see + // preupdate_callback's equivalent for why. + let _ = writeln!( + std::io::stderr(), + "sqlx-sqlite-observer: panic in commit_callback (absorbed to keep the process alive)" + ); } 0 // Allow commit to proceed } +/// Trace callback - publishes the held changes of a committed transaction. +/// +/// SQLite invokes it with the profile event when a statement finishes, and +/// with the statement event when the next statement starts. A commit that a +/// finished statement performed is durable by the time the profile event +/// fires. The statement event fires before the statement runs any of its own +/// hooks. So both events are safe publish points. Most calls find nothing +/// held and return. +/// +/// SQLite invokes the callback only for the events in its mask. The mask is +/// checked anyway, as a guard against a future registration that adds an event +/// without adding its handling here. +/// +/// Note: `user_data` is SQLite's C API term for callback context (our HookContext), +/// unrelated to application-level user data. +unsafe extern "C" fn trace_callback( + mask: c_uint, + user_data: *mut c_void, + _statement: *mut c_void, + _nanoseconds: *mut c_void, +) -> c_int { + if (mask != SQLITE_TRACE_STMT && mask != SQLITE_TRACE_PROFILE) || user_data.is_null() { + return 0; + } + + // SAFETY: user_data is a valid HookContext pointer created in register_hooks. + let context = unsafe { &*(user_data as *const HookContext) }; + + // Wrapped because an unwind out of an extern "C" fn aborts the process. + if catch_unwind(AssertUnwindSafe(|| context.publish_committed())).is_err() { + // Reported via writeln! to stderr rather than tracing - see + // preupdate_callback's equivalent for why. + let _ = writeln!( + std::io::stderr(), + "sqlx-sqlite-observer: panic in trace_callback (absorbed to keep the process alive)" + ); + } + + 0 +} + /// Rollback hook callback - discards buffered changes. /// -/// Called by SQLite when a transaction is rolled back. Fans out to every -/// broker in the map for the same reason [`commit_callback`] does - one -/// rollback hook fires for the whole transaction, not per schema. +/// Called by SQLite when a transaction is rolled back. One rollback hook fires +/// for the whole transaction, and the context holds every schema's buffered +/// events together, so one call discards them all. /// /// Note: `user_data` is SQLite's C API term for callback context (our HookContext), /// unrelated to application-level user data. @@ -434,22 +623,20 @@ unsafe extern "C" fn rollback_callback(user_data: *mut c_void) { // SAFETY: user_data is a valid HookContext pointer created in register_hooks. let context = unsafe { &*(user_data as *const HookContext) }; - // Wrapped for the same reason as commit_callback's - see its comment. - let _ = catch_unwind(|| trace!("Rollback hook fired - discarding changes")); - - // catch_unwind wraps each broker's discard individually - see - // commit_callback's comment for why the fan-out makes this matter: a - // panic partway through must not leave the remaining brokers' buffers - // undiscarded. - for broker in context.brokers.values() { - if catch_unwind(AssertUnwindSafe(|| broker.on_rollback())).is_err() { - // Reported via writeln! to stderr rather than tracing - see - // preupdate_callback's equivalent for why. - let _ = writeln!( - std::io::stderr(), - "sqlx-sqlite-observer: panic in rollback_callback (absorbed to keep the process alive)" - ); - } + + // Wrapped because an unwind out of an extern "C" fn aborts the process. + if catch_unwind(AssertUnwindSafe(|| { + trace!("Rollback hook fired - discarding changes"); + context.on_rollback(); + })) + .is_err() + { + // Reported via writeln! to stderr rather than tracing - see + // preupdate_callback's equivalent for why. + let _ = writeln!( + std::io::stderr(), + "sqlx-sqlite-observer: panic in rollback_callback (absorbed to keep the process alive)" + ); } } diff --git a/crates/sqlx-sqlite-observer/src/lib.rs b/crates/sqlx-sqlite-observer/src/lib.rs index 2fdbd2f..68c0468 100644 --- a/crates/sqlx-sqlite-observer/src/lib.rs +++ b/crates/sqlx-sqlite-observer/src/lib.rs @@ -1,7 +1,8 @@ //! Reactive change notifications for SQLite databases using sqlx. //! //! This crate provides **transaction-safe** change notifications for SQLite databases -//! using SQLite's native hooks (`preupdate_hook`, `commit_hook`, `rollback_hook`). +//! using SQLite's native hooks (`preupdate_hook`, `commit_hook`, `rollback_hook`, and a +//! `SQLITE_TRACE_PROFILE` trace hook). //! //! # SQLite Requirements //! @@ -24,7 +25,8 @@ //! //! # Features //! -//! - **Transaction-safe notifications** - changes only notify after successful commit +//! - **Transaction-safe notifications** - changes notify only once committed and +//! readable through any connection, with nothing required of the writer //! - **Typed column values** - access old/new values with native SQLite types //! - **Stream support** - use `tokio_stream::Stream` for async iteration //! - **Multiple subscribers** - broadcast channel supports multiple listeners @@ -69,7 +71,7 @@ //! .execute(&mut **conn) //! .await?; //! -//! // Changes are published automatically when the transaction commits +//! // Changes are published once the committing statement finishes //! drop(conn); //! //! Ok(()) diff --git a/crates/sqlx-sqlite-observer/src/observer.rs b/crates/sqlx-sqlite-observer/src/observer.rs index e0cf5a1..9aba704 100644 --- a/crates/sqlx-sqlite-observer/src/observer.rs +++ b/crates/sqlx-sqlite-observer/src/observer.rs @@ -20,7 +20,7 @@ use crate::schema::query_table_info; /// /// Uses SQLite's native preupdate_hook, commit_hook, and rollback_hook for /// change detection. Changes are buffered during transactions and only -/// published to subscribers after successful commit. Rolled-back transactions +/// published to subscribers once committed and readable. Rolled-back transactions /// produce no notifications. /// /// # SQLite Version Requirements @@ -54,7 +54,7 @@ impl SqliteObserver { /// /// If additional tables are provided, they will be added to the observed set. /// Returns a broadcast receiver that will receive `TableChange` events - /// after transactions commit. + /// once a committed transaction's changes are readable. pub fn subscribe(&self, tables: I) -> broadcast::Receiver where I: IntoIterator, @@ -97,8 +97,7 @@ impl SqliteObserver { /// Acquires a connection from the pool with observation hooks registered. /// - /// The returned connection will track changes to observed tables. Changes - /// are buffered during transactions and published to subscribers after commit. + /// The returned connection will track changes to observed tables. /// /// On first acquisition for each table, queries the schema to determine /// primary key columns and WITHOUT ROWID status. diff --git a/crates/sqlx-sqlite-observer/tests/conn_mgr_tests.rs b/crates/sqlx-sqlite-observer/tests/conn_mgr_tests.rs index 3d7ea7e..1aab99a 100644 --- a/crates/sqlx-sqlite-observer/tests/conn_mgr_tests.rs +++ b/crates/sqlx-sqlite-observer/tests/conn_mgr_tests.rs @@ -949,8 +949,9 @@ async fn detach_all_discards_buffered_events_for_every_broker() { .await .unwrap(); - // detach_all() calls flush_all_brokers() *before* it attempts the DETACH - - // and that flush is the only cleanup that happens here. The transaction is + // detach_all() unregisters the hooks - dropping this writer's buffer with + // them - *before* it attempts the DETACH, and that drop is the only cleanup + // that happens here. The transaction is // still open (no COMMIT/ROLLBACK was ever sent), so SQLite refuses to // detach "other" out from under it and this deterministically returns // Err(ConnMgr(Sqlx("database other is locked"))). That failure is exactly diff --git a/crates/sqlx-sqlite-observer/tests/integration_tests.rs b/crates/sqlx-sqlite-observer/tests/integration_tests.rs index 5a7c267..af6920b 100644 --- a/crates/sqlx-sqlite-observer/tests/integration_tests.rs +++ b/crates/sqlx-sqlite-observer/tests/integration_tests.rs @@ -9,7 +9,11 @@ use futures::StreamExt; use sqlx::SqlitePool; +use sqlx::sqlite::SqliteConnectOptions; use sqlx_sqlite_observer::{ChangeOperation, ColumnValue, ObserverConfig, SqliteObserver}; +use std::str::FromStr; +use std::sync::Arc; +use std::sync::atomic::{AtomicI64, Ordering}; use std::time::Duration; use tokio::time::timeout; @@ -765,3 +769,308 @@ async fn test_delete_returns_old_primary_key() { "DELETE should return old PK value" ); } + +// ============================================================================ +// Commit Visibility +// ============================================================================ + +/// A notification must mean the change is readable by another connection. +/// +/// SQLite's commit hook runs before the commit is final, so publishing from it +/// announces a change that another connection cannot read yet. This subscriber +/// reacts concurrently, reading as soon as it is notified. +/// +/// Both journal modes are covered. They make a commit visible differently: WAL +/// appends frames and publishes a new WAL index snapshot, while a rollback +/// journal writes the database file, deletes the journal, and drops the +/// EXCLUSIVE lock. Both happen inside the committing statement, before the +/// trace hook, but the branch this guards exists because a visibility +/// assumption went untested. In WAL mode an early notification shows as a +/// stale count. In rollback-journal mode it shows as a locked read, which is +/// why the reader below does not wait on the lock. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_notification_means_the_change_is_readable_in_wal_mode() { + a_notification_means_the_change_is_readable("WAL").await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_notification_means_the_change_is_readable_in_rollback_journal_mode() { + a_notification_means_the_change_is_readable("DELETE").await; +} + +async fn a_notification_means_the_change_is_readable(journal_mode: &str) { + const WRITES: i64 = 50; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("visibility.db"); + let url = format!("sqlite:{}?mode=rwc", path.display()); + + let writer_pool = SqlitePool::connect(&url).await.unwrap(); + // A test constant, not caller input; PRAGMA takes no bind parameter. + let mode: String = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "PRAGMA journal_mode={journal_mode}" + ))) + .fetch_one(&writer_pool) + .await + .unwrap(); + assert_eq!( + mode.to_uppercase(), + journal_mode, + "SQLite must accept the journal mode" + ); + sqlx::query("CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)") + .execute(&writer_pool) + .await + .unwrap(); + + // A wholly separate reader, as a consumer reacting to a change has. Its busy + // handler is off: in rollback-journal mode a reader that arrives before the + // commit released the EXCLUSIVE lock would otherwise wait for it and then + // read the committed row, which hides an early notification behind a late + // read. With no waiting, an early notification surfaces as a locked read. + let reader_options = SqliteConnectOptions::from_str(&url) + .unwrap() + .busy_timeout(Duration::ZERO); + let reader_pool = SqlitePool::connect_with(reader_options).await.unwrap(); + + let config = ObserverConfig::new().with_tables(["users"]); + let observer = SqliteObserver::new(writer_pool, config); + let mut rx = observer.subscribe(["users"]); + + let unreadable = Arc::new(AtomicI64::new(0)); + let notified = Arc::new(AtomicI64::new(0)); + + // The writer waits for the consumer to finish reading notification N before + // it starts write N+1. The race under test - commit N against read N - is + // untouched, but write N+1 can no longer hold the lock that read N runs + // into, which would count as an early notification that never happened. + let (read_done_tx, mut read_done_rx) = tokio::sync::mpsc::channel::<()>(1); + + let consumer = tokio::spawn({ + let unreadable = Arc::clone(&unreadable); + let notified = Arc::clone(¬ified); + + async move { + while rx.recv().await.is_ok() { + let count = notified.fetch_add(1, Ordering::SeqCst) + 1; + let read: Result = sqlx::query_scalar("SELECT COUNT(*) FROM users") + .fetch_one(&reader_pool) + .await; + + // A locked database means the commit had not released its lock + // when the notification arrived: unreadable, the same as a stale + // count. Any other error is a broken test, and the panic surfaces + // through the join below. + let readable = match read { + Ok(rows) => rows, + Err(sqlx::Error::Database(e)) if e.code().is_some_and(|c| c == "5") => -1, + Err(e) => panic!("reader failed: {e}"), + }; + + if readable < count { + unreadable.fetch_add(1, Ordering::SeqCst); + } + + if read_done_tx.send(()).await.is_err() { + return; + } + } + } + }); + + for i in 1..=WRITES { + let mut conn = observer.acquire().await.unwrap(); + + sqlx::query("INSERT INTO users (name) VALUES (?)") + .bind(format!("user{i}")) + .execute(&mut **conn) + .await + .unwrap(); + + // A consumer that died or never got notified stops the writes here; the + // join and the notification count below say which. + if timeout(Duration::from_secs(2), read_done_rx.recv()) + .await + .is_err() + { + break; + } + } + consumer.abort(); + + // A consumer that died on its read must fail the test with that panic, not + // with a misleading notification count. + if let Err(join_error) = consumer.await + && !join_error.is_cancelled() + { + std::panic::resume_unwind(join_error.into_panic()); + } + + assert_eq!( + notified.load(Ordering::SeqCst), + WRITES, + "every committed write must notify", + ); + assert_eq!( + unreadable.load(Ordering::SeqCst), + 0, + "a notification must not arrive before its change is readable", + ); +} + +// ============================================================================ +// Per-Connection Isolation +// ============================================================================ + +/// A transaction abandoned on one connection must not leak into another +/// connection's commit. +/// +/// Each connection's hook context owns its buffer. Connection A opens a +/// transaction, writes, and is then released mid-transaction through +/// `into_inner`, which unregisters its hooks and drops its buffer. Were the +/// buffer shared per database, A's row would still be waiting when B commits, +/// and B's commit would publish it as a phantom change. +#[tokio::test] +async fn abandoned_transaction_on_one_connection_does_not_leak_into_another() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("isolation.db"); + let url = format!("sqlite:{}?mode=rwc", path.display()); + + let pool = SqlitePool::connect(&url).await.unwrap(); + sqlx::query("CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + let config = ObserverConfig::new().with_tables(["users"]); + let observer = SqliteObserver::new(pool, config); + let mut rx = observer.subscribe(["users"]); + + let mut conn_a = observer.acquire().await.unwrap(); + let mut conn_b = observer.acquire().await.unwrap(); + + sqlx::query("BEGIN").execute(&mut **conn_a).await.unwrap(); + sqlx::query("INSERT INTO users (name) VALUES ('Pending')") + .execute(&mut **conn_a) + .await + .unwrap(); + + // Abandon A's transaction: hooks come off first, so the ROLLBACK that frees + // the write lock fires no rollback hook that could tidy up after it. + let mut raw_a = conn_a.into_inner(); + sqlx::query("ROLLBACK").execute(&mut *raw_a).await.unwrap(); + drop(raw_a); + + sqlx::query("INSERT INTO users (name) VALUES ('Real')") + .execute(&mut **conn_b) + .await + .unwrap(); + + let change = timeout(Duration::from_millis(100), rx.recv()) + .await + .expect("B's commit must notify") + .unwrap(); + assert!( + has_text_value(change.new_values.as_deref().unwrap_or_default(), "Real"), + "the one notification must be B's row, got {change:?}", + ); + + let phantom = timeout(Duration::from_millis(50), rx.recv()).await; + assert!( + phantom.is_err(), + "A's abandoned row must not surface on B's commit, got {phantom:?}", + ); +} + +// ============================================================================ +// Commits Made During a Statement Reset +// ============================================================================ + +/// A write that commits while its statement is reset must still notify, even +/// if the next statement on the connection fails. +/// +/// `fetch_one` stops at the first row and resets the statement. For an +/// autocommit `INSERT ... RETURNING`, the commit happens inside that reset. +/// SQLite fires the profile event before it halts the statement, and the halt +/// is the commit. So the trace hook sees nothing held. The change waits for +/// the next statement. If that statement fails, its rollback hook must not +/// discard the earlier, committed change. +#[tokio::test] +async fn a_commit_made_during_a_statement_reset_survives_a_later_failed_write() { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + sqlx::query( + "CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE)", + ) + .execute(&pool) + .await + .unwrap(); + + let config = ObserverConfig::new().with_tables(["users"]); + let observer = SqliteObserver::new(pool, config); + let mut rx = observer.subscribe(["users"]); + let mut conn = observer.acquire().await.unwrap(); + + let id: i64 = sqlx::query_scalar("INSERT INTO users (name) VALUES ('Alice') RETURNING id") + .fetch_one(&mut **conn) + .await + .unwrap(); + assert_eq!(id, 1); + + let duplicate = sqlx::query("INSERT INTO users (name) VALUES ('Alice')") + .execute(&mut **conn) + .await; + assert!(duplicate.is_err(), "the duplicate insert must fail"); + + let change = timeout(Duration::from_millis(100), rx.recv()) + .await + .expect("the committed insert must notify") + .unwrap(); + assert!( + has_text_value(change.new_values.as_deref().unwrap_or_default(), "Alice"), + "the notification must be the committed row, got {change:?}", + ); +} + +/// Releasing a connection publishes a commit that its statement reset made. +/// +/// Same setup as the test above, but no further statement runs. The held +/// change is published when the hooks are unregistered on drop. +#[tokio::test] +async fn dropping_a_connection_publishes_a_commit_made_during_a_statement_reset() { + let pool = SqlitePool::connect("sqlite::memory:").await.unwrap(); + sqlx::query("CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)") + .execute(&pool) + .await + .unwrap(); + + let config = ObserverConfig::new().with_tables(["users"]); + let observer = SqliteObserver::new(pool, config); + let mut rx = observer.subscribe(["users"]); + let mut conn = observer.acquire().await.unwrap(); + + let id: i64 = sqlx::query_scalar("INSERT INTO users (name) VALUES ('Alice') RETURNING id") + .fetch_one(&mut **conn) + .await + .unwrap(); + assert_eq!(id, 1); + + let before_drop = timeout(Duration::from_millis(50), rx.recv()).await; + // This assertion pins the current delivery latency. It is not a + // guarantee. If a future change publishes a reset commit earlier, update + // this assertion instead of treating the change as a regression. + assert!( + before_drop.is_err(), + "a reset commit is held until the next statement or the drop, got {before_drop:?}", + ); + + drop(conn); + + let change = timeout(Duration::from_millis(100), rx.recv()) + .await + .expect("dropping the connection must publish the held commit") + .unwrap(); + assert!( + has_text_value(change.new_values.as_deref().unwrap_or_default(), "Alice"), + "the notification must be the committed row, got {change:?}", + ); +} diff --git a/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs b/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs index 1bf9743..d005d4d 100644 --- a/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs +++ b/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs @@ -425,10 +425,12 @@ async fn concurrent_enable_observation_converges_on_one_broker() { /// A writer dropped mid-transaction - BEGIN plus a write, then dropped with no /// COMMIT or ROLLBACK ever sent - must not have its buffered change resurface -/// on the *next* transaction's commit. Without `ObservableWriteGuard::drop` -/// discarding the buffer, the abandoned INSERT below would still be sitting in -/// the broker's buffer when the real transaction commits, and `on_commit`'s -/// `mem::take` would publish it right alongside (or instead of) the real change. +/// on the *next* transaction's commit. The buffer lives in the hook context +/// that `ObservableWriteGuard::drop` unregisters, so it dies with the guard. +/// Were it to outlive the guard on the pooled connection, the abandoned INSERT +/// below would still be sitting there when the real transaction commits, and +/// the commit hook would publish it right alongside (or instead of) the real +/// change. /// /// Uses `acquire_writer()` directly (not `execute_transaction()` or /// `begin_interruptible_transaction()`) because both of those already have