diff --git a/CHANGELOG.md b/CHANGELOG.md index e05d5ed..d361a74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,9 @@ Databases must be registered on the Rust side with a stable key before they can - IPC command arguments: **`db`** → **`dbKey`**; attached **`databasePath`** → **`databaseKey`**. - **`MigrationEvent`**: adds **`dbKey`**; **`dbPath`** is now an absolute path. - **`TransactionToken`**: **`dbPath`** → **`dbKey`**. +- **`observe()` no longer resets observation.** It previously aborted every subscription for the database and rebuilt the observer; it is now additive and reference-counted (#54, see Fixed below). An app relying on that implicit reset now accumulates one **live** subscription per call, eventually failing with `TOO_MANY_SUBSCRIPTIONS`. **Migration:** `unsubscribe()` the previous subscription explicitly. +- **`subscribe()` now requires the calling window to have called `observe()` itself**, failing with `OBSERVATION_NOT_ENABLED` otherwise. Previously a window could piggyback on another window's registration, then have its subscription silently aborted when that window released it. **Migration:** every window that subscribes must call `observe()` first. +- **`observe()` now rejects a conflicting `channelCapacity`/`captureValues`** with `OBSERVATION_CONFIG_CONFLICT` instead of silently ignoring them. Both are fixed by the first window to observe a database, since the broadcast channel behind them cannot be resized without dropping subscribers. Omitting either field inherits the active value; only an explicit request for a *different* value is rejected. #### Rust API @@ -28,6 +31,8 @@ Databases must be registered on the Rust side with a stable key before they can - **`Builder::build()`** returns **`Result`**; duplicate paths across distinct registration keys fail with **`INVALID_CONFIG`**. - File paths must be **absolute** at registration; relative path resolution at load time was removed. - Invalid registration paths fail at startup (`INVALID_PATH`, `PATH_TRAVERSAL`); unregistered keys fail at open time (`PATH_NOT_REGISTERED`). +- **`DatabaseWrapper::enable_observation()` no longer tears down the existing broker** (#54, see Fixed below); it reuses one additively. Callers who re-called it to shed subscribers, or to change `channel_capacity`/`capture_values` on a live database, must now call `disable_observation()` first. It stays infallible and only logs a conflict — and that log is compiled out in release, so read back `broker().channel_capacity()` / `.capture_values()` to confirm. +- Added **`Error::ObservationConfigConflict`** (code `OBSERVATION_CONFIG_CONFLICT`). `Error` is not `#[non_exhaustive]`, so exhaustive matches on it need a new arm. ### Added @@ -48,8 +53,22 @@ Previously, `close` only aborted active subscriptions; open transactions could b Transaction cleanup failures propagate as errors rather than being logged and ignored, so a successful close indicates the database file is safe to delete or recreate. +#### `remove` aborts active transactions and is bounded by a timeout + +`Database.remove()` (IPC `plugin:sqlite|remove`) now rolls back or cancels in-flight transactions before tearing down the database's pools, and the whole teardown is bounded by the same 5-second timeout as `close`. Previously an abandoned transaction holding the write connection could make `remove()` wait indefinitely. + +`remove()` now also holds the database registry's write lock across the file deletion (see Fixed). That lock covers every loaded database, so `remove()` briefly blocks operations on unrelated databases; the timeout bounds how long. + ### Fixed +- **Re-calling `observe()` no longer terminates existing subscribers (#54).** Observation is additive and reference-counted per webview window: a second `observe()` call merges its tables into the existing broker instead of recreating it, and `unobserve()` releases only the calling window's registration. Observation is disabled once every window that called `observe()` has released. Previously any `observe()` call tore down the broker, silently ending every window's subscriptions with no error. + - `channelCapacity` and `captureValues` are fixed by the first window to enable observation, since the broadcast channel behind them cannot be resized without dropping subscribers. Omitting them inherits the active values; an explicit request for different ones is rejected with `OBSERVATION_CONFIG_CONFLICT` (see Breaking Changes). + - `observe()`, `unobserve()`, `remove()`, and the window-destroyed cleanup now hold a consistent lock order across both state stores, closing a race that could leave the broker and the recorded registrations out of sync. + - Closing a window without calling `unobserve()` no longer leaks its registration; it is released when the window is destroyed. + - **Known limitation:** the 100-observed-table limit still bounds only a single `observe()` request, not the accumulated set for a database - see the README's Resource Limits section. Because the destructive teardown was the only incremental reset of that set, a nonexistent observed table now costs schema round trips on every writer acquisition indefinitely (#56). + - **Known limitation:** observation is reference-counted per *webview*, not per caller. Two modules in the same window share one registration, so whichever calls `unobserve()` first tears down observation - and subscriptions - for both. A window needs a single owner of the `observe()`/`unobserve()` pair (#57). +- Finished subscriptions now remove their own tracking entry when their forwarding loop ends, so entries left over from a torn-down broker no longer count against the 100-subscriptions-per-database limit. **Known limitation:** this does not cover a reloaded or destroyed webview, where delivery from Rust still succeeds and the forwarding task keeps running (#58). Call `unsubscribe()` (or `unobserve()`) before navigating away or closing a window. +- `remove()` no longer deletes the database files outside the registry write lock, where a concurrent `load()` could connect to the database being torn down and then have its files unlinked underneath it - leaving the frontend writing to unlinked inodes on Unix, or failing `remove()` with the pools already closed on Windows. - Regular transaction cleanup no longer uses string-prefix matching on database keys, which could abort transactions belonging to a different registered database when keys contain `:` (for example `:memory:` or `a` vs `a:b`). - Transaction cleanup attempts all rollbacks/aborts before returning, rather than stopping at the first failure. - `close` / `close_all` now attempt pool teardown even when transaction cleanup fails, avoiding a half-closed state where subscriptions are gone but the pool remains loaded. diff --git a/Cargo.lock b/Cargo.lock index 2469ab2..e0e9c3b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3625,7 +3625,7 @@ dependencies = [ [[package]] name = "sqlx-sqlite-observer" -version = "0.9.0" +version = "0.9.1" dependencies = [ "futures", "libsqlite3-sys", @@ -3643,7 +3643,7 @@ dependencies = [ [[package]] name = "sqlx-sqlite-toolkit" -version = "0.9.0" +version = "0.9.1" dependencies = [ "base64 0.22.1", "indexmap 2.13.0", diff --git a/README.md b/README.md index 5c013bc..319eba8 100644 --- a/README.md +++ b/README.md @@ -614,7 +614,10 @@ await db.execute('INSERT INTO users (name) VALUES ($1)', ['Alice']); // 4. Unsubscribe when done await subscription.unsubscribe(); -// 5. Disable observation entirely (also aborts all active subscriptions) +// 5. Release this window's observation registration. Observation is +// reference-counted per window: this only fully disables tracking and +// aborts subscriptions once every window that called observe() for this +// database has released via unobserve(). await db.unobserve(); ``` @@ -629,8 +632,25 @@ await db.observe(['users'], { **Important:** - * Call `observe()` before `subscribe()` — subscribing without observation returns - an error + * **Every window must call `observe()` itself before `subscribe()`** — + registration is per window and is not shared, even though the broker is. + `subscribe()` enforces this, returning `OBSERVATION_NOT_ENABLED` when _this_ + window is not a registered observer + * `observe()` is additive and reference-counted: calling it again merges in the + requested tables rather than replacing anything, so existing subscriptions + keep working. `unobserve()` releases only this window's registration; tracking + stops and subscriptions abort once every registered window has released + * **Registration is keyed per _webview_, not per caller** — two modules in the + same window share one registration, so whichever calls `unobserve()` first + tears down the other's subscriptions. Treat `observe()`/`unobserve()` as owned + by a single module per window. Labels also survive a page reload without + clearing the registration, so a reloaded window can pass `subscribe()`'s check + without re-observing + * `channelCapacity`/`captureValues` are fixed by the _first_ window to observe a + database, since the broadcast channel behind them cannot be resized without + dropping subscribers. Omitting either field inherits the active value; an + explicit _different_ value fails with `OBSERVATION_CONFIG_CONFLICT`. To change + them, every window must `unobserve()` first * Multiple subscriptions can be active on the same database, each filtering by different tables * `lagged` events indicate the broadcast channel filled up before the @@ -662,7 +682,10 @@ Common error codes: * `IO_ERROR` - File system error * `MIGRATION_ERROR` - Migration failed * `MULTIPLE_ROWS_RETURNED` - `fetchOne()` returned multiple rows - * `OBSERVATION_NOT_ENABLED` - Called `subscribe()` before `observe()` + * `OBSERVATION_NOT_ENABLED` - Called `subscribe()` before this window called + `observe()` + * `OBSERVATION_CONFIG_CONFLICT` - Called `observe()` with a + `channelCapacity`/`captureValues` differing from the active one * `OBSERVER_ERROR` - Error from the observer subsystem ### Closing and Removing @@ -697,7 +720,7 @@ await db.remove(); // Close and DELETE database file(s) - irreversible | `remove()` | Close and delete database file(s), returns `true` if was loaded | | `observe(tables, config?)` | Enable change observation for tables | | `subscribe(tables, onEvent)` | Subscribe to change notifications, returns `Subscription` | -| `unobserve()` | Disable observation and abort all subscriptions | +| `unobserve()` | Release this window's observation registration; only fully disables observation and aborts subscriptions once every registered window has released | ### Builder Methods @@ -1114,7 +1137,13 @@ untrusted or buggy frontend code: default (5 minutes) are automatically rolled back on the next access attempt (configurable via `Builder::transaction_timeout()`) * **Observer channel capacity**: Capped at 10,000 (default 256) - * **Observed tables**: Maximum 100 tables per `observe()` call + * **Observed tables**: Maximum 100 tables per single `observe()` call — **not** + a bound on the accumulated set for a database. `observe()` merges its tables + into the existing broker, `subscribe()` also adds tables with no per-call + limit, and nothing removes an individual table (the set is cleared only on a + full teardown), so the total is currently unbounded. An observed table that + does not exist also costs schema round trips on _every_ writer acquisition, + indefinitely, while that database's write connection is held (#56) * **Subscriptions**: Maximum 100 active subscriptions per database ### Unbounded Result Sets diff --git a/crates/sqlx-sqlite-observer/Cargo.toml b/crates/sqlx-sqlite-observer/Cargo.toml index 877f000..9f53460 100644 --- a/crates/sqlx-sqlite-observer/Cargo.toml +++ b/crates/sqlx-sqlite-observer/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "sqlx-sqlite-observer" # Sync major.minor with major.minor of SQLx crate -version = "0.9.0" +version = "0.9.1" license = "MIT" edition = "2024" rust-version = "1.94.0" diff --git a/crates/sqlx-sqlite-observer/src/broker.rs b/crates/sqlx-sqlite-observer/src/broker.rs index 401214c..33c3556 100644 --- a/crates/sqlx-sqlite-observer/src/broker.rs +++ b/crates/sqlx-sqlite-observer/src/broker.rs @@ -55,6 +55,7 @@ pub struct ObservationBroker { observed_tables: RwLock>, table_info: RwLock>, capture_values: bool, + channel_capacity: usize, } impl ObservationBroker { @@ -76,9 +77,28 @@ impl ObservationBroker { observed_tables: RwLock::new(HashSet::new()), table_info: RwLock::new(HashMap::new()), capture_values, + channel_capacity, }) } + /// Returns the broadcast channel capacity this broker was created with. + /// + /// This is fixed for the lifetime of the broker. A broker is a long-lived object + /// that may be shared by multiple independent observers of the same database, so + /// this value cannot be changed once the broker has subscribers without dropping + /// them. + pub fn channel_capacity(&self) -> usize { + self.channel_capacity + } + + /// Returns whether this broker captures old/new column values in change notifications. + /// + /// Fixed for the lifetime of the broker; see [`channel_capacity`](Self::channel_capacity) + /// for why this can't be changed after creation. + pub fn capture_values(&self) -> bool { + self.capture_values + } + /// Checks if a table is being observed. pub fn is_table_observed(&self, table: &str) -> bool { self.observed_tables.read().contains(table) @@ -110,6 +130,24 @@ impl ObservationBroker { /// **Prefer [`observe_table`] when schema info is available**, as it atomically /// registers the table and sets schema info in one call. /// + /// This is also the merge path used when re-enabling observation on a database + /// that already has a live broker (see + /// `sqlx_sqlite_toolkit::DatabaseWrapper::enable_observation`): rather than + /// recreating the broker, callers add tables to the existing one via this + /// method, relying on `TableInfo` being filled in lazily for the newly added + /// tables on the next connection acquisition. + /// + /// That lazy fill only converges for tables that actually exist. For a name + /// that is not in the schema, `query_table_info` yields `Ok(None)` and nothing + /// is recorded, so the name stays in the "needs querying" set and costs two + /// statements on *every* `acquire_writer()` for the lifetime of the broker. + /// Changes are still delivered if the table is created later - the hook gates + /// on the observed set, not on `TableInfo` - but until the info resolves those + /// events carry an empty `primary_key`, and a meaningless `rowid` for a + /// `WITHOUT ROWID` table. Nothing removes an individual name from the observed + /// set, so a typo or a table dropped by a later migration keeps paying that + /// cost indefinitely. + /// /// [`set_table_info`]: Self::set_table_info /// [`observe_table`]: Self::observe_table pub fn observe_tables(&self, tables: I) @@ -301,6 +339,41 @@ impl std::fmt::Debug for ObservationBroker { 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) .finish() } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_channel_capacity_and_capture_values_are_fixed_after_creation() { + let broker = ObservationBroker::new(64, false); + + assert_eq!(broker.channel_capacity(), 64); + assert!(!broker.capture_values()); + } + + #[test] + fn test_observe_tables_is_additive_and_idempotent() { + let broker = ObservationBroker::new(16, true); + + broker.observe_tables(["users"]); + assert_eq!(broker.get_observed_tables(), vec!["users".to_string()]); + + // Adding more tables unions them with the existing set rather than replacing it. + broker.observe_tables(["posts"]); + let mut observed = broker.get_observed_tables(); + observed.sort(); + assert_eq!(observed, vec!["posts".to_string(), "users".to_string()]); + + // Re-observing an already-tracked table is a no-op, not a duplicate. + broker.observe_tables(["users"]); + let mut observed = broker.get_observed_tables(); + observed.sort(); + assert_eq!(observed, vec!["posts".to_string(), "users".to_string()]); + } +} diff --git a/crates/sqlx-sqlite-toolkit/Cargo.toml b/crates/sqlx-sqlite-toolkit/Cargo.toml index 9170a53..81b5712 100644 --- a/crates/sqlx-sqlite-toolkit/Cargo.toml +++ b/crates/sqlx-sqlite-toolkit/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "sqlx-sqlite-toolkit" # Sync major.minor with major.minor of SQLx crate -version = "0.9.0" +version = "0.9.1" license = "MIT" edition = "2024" rust-version = "1.94.0" diff --git a/crates/sqlx-sqlite-toolkit/src/wrapper.rs b/crates/sqlx-sqlite-toolkit/src/wrapper.rs index 9fb1f69..aea9b4d 100644 --- a/crates/sqlx-sqlite-toolkit/src/wrapper.rs +++ b/crates/sqlx-sqlite-toolkit/src/wrapper.rs @@ -5,6 +5,8 @@ use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; use sqlx::sqlite::SqliteConnection; use sqlx_sqlite_conn_mgr::{SqliteDatabase, SqliteDatabaseConfig, WriteGuard}; +#[cfg(feature = "observer")] +use tracing::warn; #[cfg(feature = "observer")] use sqlx_sqlite_observer::{ObservableSqliteDatabase, ObservableWriteGuard, ObserverConfig}; @@ -387,14 +389,60 @@ impl DatabaseWrapper { /// After calling this, write operations will be tracked and subscribers /// can receive change notifications. /// - /// If observation is already enabled, the previous observer is disabled first. - /// This drops the old broadcast broker, causing existing subscriber streams to - /// terminate. Callers must re-subscribe after re-enabling observation. + /// **Additive, not destructive:** if observation is already enabled, the existing + /// broker is reused rather than replaced. The requested tables are unioned into + /// its observed-table set, and any subscribers created before this call keep + /// receiving notifications uninterrupted — this is what allows independent callers + /// (e.g. multiple windows observing the same database) to call `enable_observation` + /// without tearing down each other's subscriptions. + /// + /// `config.channel_capacity` and `config.capture_values` can only take effect on + /// the *first* call that enables observation for this database. Both are baked + /// into the broadcast channel/broker at creation time and cannot be changed + /// without recreating the broker — which would drop existing subscribers, the + /// exact problem this method now avoids. If a later call requests different + /// values, they are ignored (a warning is logged) and only the tables are merged + /// in — this method stays infallible on purpose. The Tauri plugin layer's + /// `observe()` command rejects a conflicting request outright before it ever + /// reaches here, so in practice this fallback only matters for direct Rust + /// callers of this crate. Those callers should not rely on the warning to + /// notice: all four crates in this workspace pin `tracing` with + /// `release_max_level_off`, which compiles the `warn!` below out entirely + /// whenever `debug_assertions` are disabled — not merely when the build + /// profile happens to be named `release`. The only reliable way to learn the values actually in effect + /// is to read them back afterward via `broker().channel_capacity()` / + /// `.capture_values()`. Call [`disable_observation`](Self::disable_observation) + /// first if you need to change these values, accepting that existing + /// subscribers will be dropped. /// /// Requires the `observer` feature. #[cfg(feature = "observer")] pub fn enable_observation(&mut self, config: ObserverConfig) { - self.disable_observation(); + if let Some(existing) = &self.observer { + let broker = existing.broker(); + + if config.channel_capacity != broker.channel_capacity() + || config.capture_values != broker.capture_values() + { + warn!( + requested_channel_capacity = config.channel_capacity, + active_channel_capacity = broker.channel_capacity(), + requested_capture_values = config.capture_values, + active_capture_values = broker.capture_values(), + "enable_observation() called with different channel_capacity/capture_values \ + while observation is already active; keeping the original values since \ + recreating the broadcast channel would drop existing subscribers. Only the \ + requested tables were merged in." + ); + } + + if !config.tables.is_empty() { + broker.observe_tables(config.tables.iter().map(String::as_str)); + } + + return; + } + self.observer = Some(ObservableSqliteDatabase::new( Arc::clone(&self.inner), config, diff --git a/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs b/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs new file mode 100644 index 0000000..b264b59 --- /dev/null +++ b/crates/sqlx-sqlite-toolkit/tests/observation_tests.rs @@ -0,0 +1,147 @@ +//! Regression tests for `DatabaseWrapper::enable_observation()`. +//! +//! These specifically cover issue #54: re-calling `enable_observation()` (surfaced +//! to Tauri callers as `observe()`) must not destroy the existing broadcast broker, +//! or every subscriber created before the re-call silently stops receiving events. + +#![cfg(feature = "observer")] + +use std::time::Duration; + +use sqlx_sqlite_observer::ObserverConfig; +use sqlx_sqlite_toolkit::DatabaseWrapper; +use tempfile::TempDir; +use tokio::time::timeout; + +async fn create_test_db() -> (DatabaseWrapper, TempDir) { + let temp_dir = TempDir::new().expect("Failed to create temp directory"); + let db_path = temp_dir.path().join("test.db"); + let wrapper = DatabaseWrapper::connect(&db_path, None) + .await + .expect("Failed to connect to test database"); + + wrapper + .execute( + "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)".into(), + vec![], + ) + .await + .expect("create users table"); + wrapper + .execute( + "CREATE TABLE posts (id INTEGER PRIMARY KEY, title TEXT)".into(), + vec![], + ) + .await + .expect("create posts table"); + + (wrapper, temp_dir) +} + +#[tokio::test] +async fn test_first_enable_observation_applies_requested_config() { + let (mut wrapper, _temp) = create_test_db().await; + + wrapper.enable_observation( + ObserverConfig::new() + .with_tables(["users"]) + .with_channel_capacity(8) + .with_capture_values(false), + ); + + let broker = wrapper.observable().unwrap().broker(); + assert_eq!(broker.channel_capacity(), 8); + assert!(!broker.capture_values()); + assert!(wrapper.is_observing()); +} + +/// This is the exact regression scenario from issue #54: a subscriber created +/// before a second `enable_observation()` call (with a *different* table set) +/// must keep receiving events published after that second call, instead of +/// seeing its `broadcast::Receiver` closed because the broker was replaced. +#[tokio::test] +async fn test_reenable_observation_preserves_existing_subscriber_across_new_tables() { + let (mut wrapper, _temp) = create_test_db().await; + + wrapper.enable_observation(ObserverConfig::new().with_tables(["users"])); + + // A subscriber that only asked to observe "users" tables, created before the + // second (additive) enable_observation() call below. + let mut rx = wrapper.observable().unwrap().subscribe(["users"]); + + // Second observe() call, with a completely different table set. Under the old + // (destructive) behavior this would drop the broker and close `rx`. + wrapper.enable_observation(ObserverConfig::new().with_tables(["posts"])); + + wrapper + .execute("INSERT INTO users (name) VALUES ('Alice')".into(), vec![]) + .await + .expect("insert into users"); + + let change = timeout(Duration::from_millis(200), rx.recv()) + .await + .expect("subscriber should not have been closed by the second enable_observation() call") + .expect("should receive a change, not RecvError::Closed"); + + assert_eq!(change.table, "users"); + + // The second call's tables were merged in, not swapped in. + let mut observed = wrapper.observable().unwrap().observed_tables(); + observed.sort(); + assert_eq!(observed, vec!["posts".to_string(), "users".to_string()]); +} + +/// Config-conflict rule: `channel_capacity` and `capture_values` are fixed by the +/// first `enable_observation()` call. A later call requesting different values for +/// either is ignored for those two fields (only the tables are merged in) - this is +/// the direct-Rust-caller contract for this crate. The Tauri plugin's `observe()` +/// command builds a stricter contract on top of it (rejecting the conflicting +/// request outright), but that's enforced one layer up in `src/commands.rs` of the +/// `tauri-plugin-sqlite` crate, not here. +#[tokio::test] +async fn test_reenable_observation_ignores_conflicting_config() { + let (mut wrapper, _temp) = create_test_db().await; + + wrapper.enable_observation( + ObserverConfig::new() + .with_tables(["users"]) + .with_channel_capacity(4) + .with_capture_values(false), + ); + + wrapper.enable_observation( + ObserverConfig::new() + .with_tables(["users"]) + .with_channel_capacity(999) + .with_capture_values(true), + ); + + let broker = wrapper.observable().unwrap().broker(); + assert_eq!( + broker.channel_capacity(), + 4, + "channel_capacity should stay at the first call's value" + ); + assert!( + !broker.capture_values(), + "capture_values should stay at the first call's value" + ); + + // Confirm the ignored capture_values request is reflected in actual behavior, + // not just the getter: old/new values should still be absent from change events. + let mut rx = wrapper.observable().unwrap().subscribe(["users"]); + wrapper + .execute("INSERT INTO users (name) VALUES ('Bob')".into(), vec![]) + .await + .expect("insert into users"); + + let change = timeout(Duration::from_millis(200), rx.recv()) + .await + .expect("should not time out") + .expect("should receive a change"); + + assert!( + change.new_values.is_none(), + "capture_values=true from the second call should have been ignored" + ); +} diff --git a/guest-js/index.ts b/guest-js/index.ts index a205a21..733d204 100644 --- a/guest-js/index.ts +++ b/guest-js/index.ts @@ -313,6 +313,11 @@ export interface KeysetPage> { /** * Configuration for the database observer. + * + * `channelCapacity` and `captureValues` are fixed by the first window to + * enable observation for a given database. Omitting this config on a later + * `observe()` call always succeeds, but an explicit request for different + * values for either field is rejected. */ export interface ObserverConfig { @@ -1192,13 +1197,31 @@ export default class Database { * Must be called before `subscribe()`. This configures the database to track * changes via SQLite hooks. Changes are only published after transactions commit. * - * If observation is already enabled, calling this again will abort all existing - * subscriptions for this database, tear down the previous observer, and create - * a new one with the provided configuration. You must re-subscribe after - * re-calling `observe()`. + * Observation is additive and reference-counted per window: calling this again + * (from this window or another one) merges the requested tables into the + * existing observer rather than replacing it, so subscriptions already active + * in any window - including this one - keep receiving notifications + * uninterrupted. `channelCapacity` and `captureValues` are fixed by the + * *first* window to enable observation for a given database; a later call + * omitting them always succeeds, but one requesting different values for + * either is rejected. + * + * Registration is keyed by webview label, not by caller: if two independent + * modules in the same window both call `observe()`, they share a single + * registration, and whichever one calls `unobserve()` first tears down + * observation - and aborts subscriptions - for both. Treat `observe()` / + * `unobserve()` as owned by a single module per window. + * + * Call `unobserve()` to release this window's registration. The underlying + * observer is only disabled once every window that called `observe()` for + * this database has released its registration. * * @param tables - Table names to observe for changes * @param config - Optional observer configuration + * @throws SqliteError with code `OBSERVATION_CONFIG_CONFLICT` if `config` + * requests a `channelCapacity`/`captureValues` different from the + * values a prior `observe()` call already established for this + * database * * @example * ```ts @@ -1229,11 +1252,24 @@ export default class Database { * Returns a `Subscription` that can be used to unsubscribe later. Change events * are streamed to the provided callback function. * - * Requires `observe()` to have been called first. + * The calling window must have called `observe()` itself - another window + * having called `observe()` does not satisfy this requirement. + * + * Because registration is keyed by webview label rather than by caller, a + * subscription can also be aborted if another module in the same window + * calls `unobserve()`, even if it wasn't the module that called `observe()`. + * See `observe()` for details. + * + * @remarks + * Webview labels persist across a page reload and registrations are not + * cleared on reload, so a window may pass the `observe()` check after a + * reload without having re-called `observe()` in the new page load. * * @param tables - Table names to receive notifications for * @param onEvent - Callback invoked for each change event * @returns A Subscription that can be used to stop receiving notifications + * @throws SqliteError with code `OBSERVATION_NOT_ENABLED` if this window + * has not itself called `observe()` for this database * * @example * ```ts @@ -1274,9 +1310,13 @@ export default class Database { /** * **unobserve** * - * Disable change observation for this database. + * Release this window's change observation registration for this database. * - * Stops tracking changes and aborts all active subscriptions for this database. + * Observation is reference-counted per window: if other windows are still + * observing this database, this only removes this window's own registration - + * tracking stays enabled and every other window's subscriptions are left + * untouched. Only once the *last* registered window calls `unobserve()` are + * changes actually stopped and all subscriptions for this database aborted. * * @example * ```ts diff --git a/src/commands.rs b/src/commands.rs index fac1298..817d439 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -2,6 +2,81 @@ //! //! This module implements the Tauri command handlers that the frontend calls. //! Each command manages database connections through the DbInstances state. +//! +//! # Lock order: `DbInstances` before `ObserverRegistrations` +//! +//! Any command that touches both the `db_instances` map (which owns each +//! database's observer/broker via `DatabaseWrapper`) and `observer_regs` (the +//! per-webview refcount tracking who is observing each database) must acquire +//! `db_instances.inner`'s write lock first, then perform its `observer_regs` +//! register/release call while still holding it - not release the db lock and +//! re-acquire `observer_regs` afterward. `observe()`, `unobserve()`, +//! `remove()`'s `remove_inner` helper, `close_database_inner`, +//! `close_all_wrappers`, and the window-`Destroyed` cleanup handler all follow +//! this order today. +//! +//! ## What the compiler checks, and what it doesn't +//! +//! This is more than a documented convention, but less than a proof. Every +//! mutating method on `ObserverRegistrations` (`register`, `release`, +//! `release_all_for_label`, `clear_for_db`, `clear_all`) takes a +//! [`DbInstancesGuard`](crate::subscriptions::DbInstancesGuard) as its first +//! parameter - a witness that the caller holds a `DbInstances` write guard at +//! the point of the call. So *co-holding* the two locks is a compile-time +//! obligation: a call site with no db guard in scope fails to compile. That is +//! the shape that was reintroduced four separate times while the rule lived +//! only in prose here. +//! +//! Two things the type system structurally cannot see: +//! +//! - **Acquisition order.** A witness proves a guard exists, not that it was +//! taken before `observer_regs`'s lock. +//! - **Drop and reacquire mid-sequence.** Releasing the db guard partway +//! through and immediately taking a fresh one satisfies the witness (a real +//! guard is a real guard) and passes both deterministic tests, while +//! reopening the exact race this rule exists to close. +//! +//! Those shapes are covered by tests only, and unevenly. +//! `tests::test_observe_holds_db_lock_across_register` and +//! `tests::test_unobserve_holds_db_lock_across_release` in `src/lib.rs` are +//! deterministic and authoritative for "a guard is held at the call site", one +//! per side. `tests::test_concurrent_observe_and_unobserve_keep_broker_and_registrations_in_sync` +//! is the *only* guard for the drop-and-reacquire shape, and it is +//! probabilistic - its detection rate for a regressed `observe()` side is low +//! (see that test's doc). Treat a green CI run as evidence rather than a +//! guarantee here, and read that doc before changing it. +//! +//! Without a single consistent order held across the whole +//! enable+register/release+disable sequence, two commands running +//! concurrently for the same database (e.g. window B's `observe()` racing +//! window A's `unobserve()`) can interleave such that `is_observing()` and +//! "has any registered observers" disagree - e.g. B registers into a broker +//! that A's concurrent `unobserve()` just destroyed after a stale refcount +//! read. The tests named above are the regression guards. +//! +//! ## `ActiveSubscriptions` is a separate pair, with no fixed order +//! +//! The rule above governs the `db_instances` / `observer_regs` pair only. A +//! third store, `active_subs`, is locked in *both* orders relative to the db +//! lock: `unobserve()` and the window-`Destroyed` handler in `src/lib.rs` call +//! `active_subs.remove_for_db()` while still holding the db guard, whereas +//! `remove()`, `close_database_inner` and `close_all_loaded_databases` take and +//! release `active_subs` *before* acquiring it. `subscribe()` is in both +//! orderings at once: it calls `active_subs.count_for_db()` before acquiring +//! the db lock, then `active_subs.insert()` while still holding it. Do not +//! "fix" one side to match the other by moving a call across a db-lock +//! acquisition without re-reading the reason below. +//! +//! That inconsistency cannot deadlock, and the reason is checkable rather than +//! a matter of inspection: no `ActiveSubscriptions` method ever acquires +//! another lock while holding its own. Everything it does under that lock is a +//! map removal, a `String` comparison, and `AbortHandle::abort()` - which +//! schedules task shutdown on the runtime instead of running the aborted +//! future's destructor inline. So the "wrong" order never closes a cycle: one +//! side waits on `active_subs` while holding the db lock, but nothing ever +//! waits on the db lock while holding `active_subs`. That property follows from +//! what `ActiveSubscription` stores - see its doc comment - not from the call +//! order here, so a change to that struct is what could break it. use futures::StreamExt; use indexmap::IndexMap; @@ -21,7 +96,8 @@ use uuid::Uuid; use crate::{ DbInstances, Error, MigrationEvent, MigrationStates, Result, subscriptions::{ - ActiveSubscriptions, ObserverConfigParams, TableChangePayload, event_to_payload, + ActiveSubscriptions, ObserverConfigParams, ObserverRegistrations, TableChangePayload, + event_to_payload, }, }; use crate::{close_all_loaded_databases, close_database, connect_to_database}; @@ -338,6 +414,7 @@ pub async fn fetch_page( pub async fn close( db_instances: State<'_, DbInstances>, active_subs: State<'_, ActiveSubscriptions>, + observer_regs: State<'_, ObserverRegistrations>, interruptible_txs: State<'_, ActiveInterruptibleTransactions>, regular_txs: State<'_, ActiveRegularTransactions>, db_key: String, @@ -346,6 +423,7 @@ pub async fn close( &db_key, &db_instances, &active_subs, + &observer_regs, &interruptible_txs, ®ular_txs, ) @@ -362,12 +440,14 @@ pub async fn close( pub async fn close_all( db_instances: State<'_, DbInstances>, active_subs: State<'_, ActiveSubscriptions>, + observer_regs: State<'_, ObserverRegistrations>, interruptible_txs: State<'_, ActiveInterruptibleTransactions>, regular_txs: State<'_, ActiveRegularTransactions>, ) -> Result<()> { close_all_loaded_databases( &db_instances, &active_subs, + &observer_regs, &interruptible_txs, ®ular_txs, ) @@ -378,22 +458,99 @@ pub async fn close_all( /// /// Returns `true` if the database was loaded and successfully removed. /// Returns `false` if the database was not loaded (nothing to remove). -/// Any active subscriptions for this database are aborted before removing. +/// Returns `Err` if transaction cleanup or file removal fails, or if the +/// whole operation doesn't finish within `CLOSE_TIMEOUT` (see `remove_inner` +/// for why that bound exists here and not just on `close()`/`close_all()`). +/// Active subscriptions for this key are aborted, and in-flight transactions +/// are cleaned up (interruptible transactions rolled back; regular +/// transaction tasks aborted and awaited) before the connection pool is +/// closed and the database's files are deleted. #[tauri::command] pub async fn remove( db_instances: State<'_, DbInstances>, active_subs: State<'_, ActiveSubscriptions>, + observer_regs: State<'_, ObserverRegistrations>, + interruptible_txs: State<'_, ActiveInterruptibleTransactions>, + regular_txs: State<'_, ActiveRegularTransactions>, db_key: String, ) -> Result { - active_subs.remove_for_db(&db_key).await; + let remove_result = tokio::time::timeout( + crate::CLOSE_TIMEOUT, + remove_inner( + &db_instances, + &active_subs, + &observer_regs, + &interruptible_txs, + ®ular_txs, + &db_key, + ), + ) + .await; + + match remove_result { + Ok(result) => result, + Err(_) => Err(Error::Other(format!( + "database remove timed out after {} seconds", + crate::CLOSE_TIMEOUT.as_secs() + ))), + } +} - let mut instances = db_instances.inner.write().await; +/// Abort in-flight transactions and subscriptions for `db_key`, then remove +/// its wrapper and delete its files - attempting the removal even if +/// transaction cleanup failed, mirroring `close_database_inner`'s +/// best-effort teardown in `src/lib.rs`. +/// +/// Transaction cleanup must run first. `begin_interruptible_transaction` checks +/// the write connection out for the transaction's whole lifetime, and an +/// abandoned one is only reaped lazily, so without this call +/// `wrapper.remove()`'s `Pool::close()` - which has no timeout of its own - +/// waits on that connection indefinitely. +/// +/// Lock order: db_instances write lock, then observer_regs (see the module doc). +/// One guard covers the map removal, the registration clear, and +/// `wrapper.remove()` itself, so nothing can interleave between them. That last +/// part matters because `wrapper.remove()` unlinks the `.db`/`-wal`/`-shm` +/// files: release the guard any earlier and a concurrent `connect_to_database()` +/// can hand back a live handle to this database - the wrapper being torn down, +/// or a freshly connected one at the same path - which the unlink then deletes +/// the files out from under. +/// +/// That isn't free. `db_instances`'s write lock covers every loaded database, so +/// holding it across pool teardown and file I/O stalls unrelated databases; the +/// `CLOSE_TIMEOUT` in `remove()` bounds that rather than leaving it open-ended. +async fn remove_inner( + db_instances: &DbInstances, + active_subs: &ActiveSubscriptions, + observer_regs: &ObserverRegistrations, + interruptible_txs: &ActiveInterruptibleTransactions, + regular_txs: &ActiveRegularTransactions, + db_key: &str, +) -> Result { + let mut last_error = None; - if let Some(wrapper) = instances.remove(&db_key) { - wrapper.remove().await?; - Ok(true) - } else { - Ok(false) // Database wasn't loaded + active_subs.remove_for_db(db_key).await; + + if let Err(err) = + sqlx_sqlite_toolkit::cleanup_transactions_for_db(db_key, interruptible_txs, regular_txs).await + { + last_error = Some(err.into()); + } + + let mut instances = db_instances.write().await; + let wrapper = instances.remove(db_key); + observer_regs.clear_for_db(&mut instances, db_key).await; + + let was_loaded = wrapper.is_some(); + if let Some(wrapper) = wrapper + && let Err(err) = wrapper.remove().await + { + last_error = Some(err.into()); + } + + match last_error { + Some(err) => Err(err), + None => Ok(was_loaded), } } @@ -566,13 +723,33 @@ pub async fn transaction_read( /// Must be called before `subscribe()`. Configures the observer with the /// specified tables and options. /// -/// If observation is already enabled, this will abort all existing subscriptions -/// for this database, disable the previous observer, and enable a new one with -/// the provided configuration. Callers must re-subscribe after re-calling this. +/// Observation is additive and reference-counted per webview: calling this again +/// (from the same or a different window) merges the requested tables into the +/// existing broker rather than replacing it, so subscriptions already active in +/// any window - including this one - keep receiving notifications uninterrupted. +/// See issue #54. +/// +/// `channelCapacity` and `captureValues` can only be set by the *first* window to +/// enable observation for a given database; a later call explicitly requesting +/// different values for either is rejected with `OBSERVATION_CONFIG_CONFLICT` +/// (see `sqlx_sqlite_toolkit::DatabaseWrapper::enable_observation` for the +/// underlying rule and why it can't be changed without dropping existing +/// subscribers). Omit the conflicting field(s) to keep using the active value, +/// or have every window call `unobserve()` first if the value must change. +/// +/// `MAX_OBSERVED_TABLES` only bounds the size of a single `observe()` call's +/// request, not the accumulated set of tables observed on a database overall - +/// see the Resource Limits section of the README for why that's intentionally +/// left unbounded for now rather than fixed the wrong way. +/// +/// Call `unobserve()` to release this window's registration. The underlying +/// broker and its subscriptions are only torn down once every window that +/// called `observe()` for this database has released its registration. #[tauri::command] -pub async fn observe( +pub async fn observe( db_instances: State<'_, DbInstances>, - active_subs: State<'_, ActiveSubscriptions>, + observer_regs: State<'_, ObserverRegistrations>, + webview: tauri::Webview, db_key: String, tables: Vec, config: Option, @@ -587,17 +764,38 @@ pub async fn observe( ))); } - // Abort plugin-level subscription tasks before the crate-level - // enable_observation() drops the old broker - active_subs.remove_for_db(&db_key).await; - - let mut instances = db_instances.inner.write().await; + // Lock order: db_instances write lock, then observer_regs - matched exactly + // in unobserve() below, and held across the *entire* enable+register + // sequence. Without a single lock spanning both state stores, a concurrent + // unobserve() from another window could see this window's registration land + // after it already decided the refcount had hit zero and torn the broker + // down - registering into a broker that no longer exists, while observe() + // had already returned Ok(()) to the frontend. + let mut instances = db_instances.write().await; let wrapper = instances .get_mut(&db_key) .ok_or_else(|| Error::DatabaseNotLoaded(db_key.clone()))?; + // Seed channelCapacity/captureValues from the live broker (if one exists) + // rather than ObserverConfig's hardcoded defaults. Otherwise a + // caller that supplies no `config` at all still produces a fully-populated + // ObserverConfig with the crate's defaults (256 / true), which would look + // like a genuinely conflicting request - and spuriously warn - the moment + // another window's broker is already using different values. + let seeded = wrapper.observable().map(|existing| { + ( + existing.broker().channel_capacity(), + existing.broker().capture_values(), + ) + }); + let mut observer_config = sqlx_sqlite_observer::ObserverConfig::new().with_tables(tables); + if let Some((capacity, capture)) = seeded { + observer_config = observer_config + .with_channel_capacity(capacity) + .with_capture_values(capture); + } if let Some(params) = config { if let Some(capacity) = params.channel_capacity { @@ -613,7 +811,54 @@ pub async fn observe( } } + // Reject rather than silently ignore an explicit request that conflicts with + // the live broker's already-fixed channelCapacity/captureValues - both are + // baked into the broadcast channel at creation time and can't change without + // dropping every existing subscriber (see `enable_observation`'s doc for why). + // Comparing the final `observer_config` against `seeded` is exactly "an + // explicit param differs from the live broker": `observer_config` only + // diverges from `seeded` above when a `params` field explicitly overrode it, + // since a caller that omits a field never moves it off the seeded value. + if let Some((capacity, capture)) = seeded { + if observer_config.channel_capacity != capacity { + return Err(Error::ObservationConfigConflict(format!( + "observe() requested channelCapacity {} for database {db_key}, but \ + observation is already active with channelCapacity {capacity}; this value \ + is fixed by the first window to enable observation and can't be changed \ + without dropping every existing subscriber. Omit channelCapacity to keep \ + using the active value, or have every window call unobserve() first if it \ + must change.", + observer_config.channel_capacity + ))); + } + if observer_config.capture_values != capture { + return Err(Error::ObservationConfigConflict(format!( + "observe() requested captureValues {} for database {db_key}, but observation \ + is already active with captureValues {capture}; this value is fixed by the \ + first window to enable observation and can't be changed without dropping \ + every existing subscriber. Omit captureValues to keep using the active \ + value, or have every window call unobserve() first if it must change.", + observer_config.capture_values + ))); + } + } + + // Additive: reuses the existing broker (if any) rather than tearing it down, + // so subscriptions belonging to other windows are never disturbed here. wrapper.enable_observation(observer_config); + + // Track this window as an observer of db_key so unobserve() knows whether + // it's safe to tear the broker down (see ObserverRegistrations docs). + // Registered while still holding `instances` - see the lock-order comment + // above. + let observer_count = observer_regs + .register(&mut instances, &db_key, webview.label()) + .await; + debug!( + "observe: database {} now has {} registered observer(s)", + db_key, observer_count + ); + Ok(()) } @@ -622,11 +867,21 @@ pub async fn observe( /// Returns a subscription ID that can be used to unsubscribe later. /// Change events are streamed to the frontend via Tauri Channel. /// -/// Requires `observe()` to have been called first. +/// The calling window must have called `observe()` for `db_key` itself first; +/// this returns `OBSERVATION_NOT_ENABLED` otherwise, even if some other +/// window's registration currently has a broker active for this database (see +/// issue #54 - piggybacking on another window's registration used to let a +/// subscription outlive its own observer, then be silently aborted the moment +/// that other window called `unobserve()`). One caveat survives: webview +/// labels persist across a reload and registrations aren't cleared on one, so +/// this proves "this webview label called `observe()` at some point", not +/// "this specific page load did". #[tauri::command] -pub async fn subscribe( +pub async fn subscribe( db_instances: State<'_, DbInstances>, active_subs: State<'_, ActiveSubscriptions>, + observer_regs: State<'_, ObserverRegistrations>, + webview: tauri::Webview, db_key: String, tables: Vec, on_event: Channel, @@ -648,6 +903,19 @@ pub async fn subscribe( .observable() .ok_or_else(|| Error::ObservationNotEnabled(db_key.clone()))?; + // This webview must be one of db_key's registered observers itself, not + // merely subscribing while *some* window's broker happens to exist (see the + // doc comment above and issue #54). Checked while still holding + // `instances`'s read lock, alongside the `observable()` check above: + // `unobserve()` needs the write lock to release a registration and tear the + // broker down, so holding this read lock across the check and + // `subscribe_stream()` below is what prevents a concurrent `unobserve()` + // from doing so in between - the same lock order documented at the top of + // this file. + if !observer_regs.is_registered(&db_key, webview.label()).await { + return Err(Error::ObservationNotEnabled(db_key.clone())); + } + // Create subscription stream let mut stream = observable.subscribe_stream(tables); @@ -657,8 +925,22 @@ pub async fn subscribe( // Spawn task to forward stream events to the Tauri Channel let sub_id = subscription_id.clone(); let db_key_clone = db_key.clone(); + let active_subs_for_task = active_subs.inner().clone(); + + // Ready signal so the task can't start forwarding (and, at the end, reaping + // its own entry) until this function has actually inserted that entry into + // `active_subs` below - otherwise, on a multi-thread runtime, a task whose + // stream/channel ends immediately could run to completion and call + // `remove()` on another worker thread *before* `insert()` below has run, + // leaving a since-finished subscription registered forever with no task to + // ever reap it - the exact leak the reap at the end of the task prevents. + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<()>(); let handle = tokio::spawn(async move { + // See the ready_tx/ready_rx comment above: wait until this subscription + // is registered in `active_subs` before doing anything observable. + let _ = ready_rx.await; + while let Some(event) = stream.next().await { let payload = event_to_payload(event); if on_event.send(payload).is_err() { @@ -669,12 +951,31 @@ pub async fn subscribe( } debug!("Subscription {} for db {} ended", sub_id, &db_key_clone); + + // Reap this subscription's own entry now that its forwarding loop has + // ended. This does NOT cover a same-webview reload: `on_event.send()` + // resolves to `Channel::send`, which ends in `webview.eval(...)`, and in + // a default build (`tracing` isn't a default Tauri feature) that's + // `tauri-runtime-wry`'s `send_user_message` variant - it returns `Ok` as + // soon as the message is queued to the app-global event-loop proxy, and + // the handler is a no-op if the webview it's addressed to is gone. So + // The loop above only breaks when the upstream broker/stream ends or the + // event loop tears down: `send()` returns `Ok` for a reloaded webview + // (still alive, its JS callback just isn't listening) and for a destroyed + // one alike, so neither triggers this. Reaping here stops a finished + // subscription from leaving an entry that counts against + // MAX_SUBSCRIPTIONS_PER_DATABASE forever, which would eventually fail + // every new subscribe() with TOO_MANY_SUBSCRIPTIONS while nothing is + // receiving events. Safe to race with unsubscribe() - `remove()` and + // aborting an already-finished `AbortHandle` are both no-ops. + active_subs_for_task.remove(&sub_id).await; }); - // Track subscription + // Track subscription, then release the task to start forwarding/reaping. active_subs .insert(subscription_id.clone(), db_key, handle.abort_handle()) .await; + let _ = ready_tx.send(()); Ok(subscription_id) } @@ -690,24 +991,78 @@ pub async fn unsubscribe( Ok(active_subs.remove(&subscription_id).await) } -/// Disable observation on a database. +/// Release this window's observation registration for a database. +/// +/// Observation is reference-counted per webview (see `observe()` docs and issue +/// #54): if other windows are still observing this database, this call only +/// removes this window's own registration and returns - the broker and every +/// other window's subscriptions are left untouched. Only when the *last* +/// registered window calls `unobserve()` are changes actually stopped, tracking +/// disabled, and all subscriptions for this database aborted. /// -/// Stops tracking changes and aborts all subscriptions for this database. +/// Calling this from a window that never called `observe()` for `db_key` is a +/// no-op (beyond validating that `db_key` itself is loaded) - it does not tear +/// down observation that other windows are legitimately still using. #[tauri::command] -pub async fn unobserve( +pub async fn unobserve( db_instances: State<'_, DbInstances>, active_subs: State<'_, ActiveSubscriptions>, + observer_regs: State<'_, ObserverRegistrations>, + webview: tauri::Webview, db_key: String, ) -> Result<()> { - // Abort all subscriptions for this database first - active_subs.remove_for_db(&db_key).await; + // Lock order matches observe(): db_instances write lock first, then + // observer_regs, held across the entire release+disable sequence. + // Acquiring the db lock unconditionally, before knowing the refcount, + // also fixes an inconsistency a caller could otherwise observe: this used + // to only validate `db_key` is loaded on the "last observer" path, so an + // unloaded/unregistered `db_key` combined with a non-zero remaining count + // would silently succeed instead of erroring like every other command does. + let mut instances = db_instances.write().await; + + if !instances.contains_key(&db_key) { + return Err(Error::DatabaseNotLoaded(db_key.clone())); + } - let mut instances = db_instances.inner.write().await; + // `release()` takes `&mut instances` as a witness that we're still holding + // the db lock (see `ObserverRegistrations`'s lock-order doc) - it doesn't + // read through it, so this doesn't conflict with re-borrowing `instances` + // below to fetch the wrapper. We can't hold that wrapper borrow across this + // call instead, because the borrow checker won't allow a live `&mut + // DatabaseWrapper` (from `instances.get_mut()`) at the same time as this + // `&mut instances` - which is exactly the kind of thing the witness + // parameter is meant to force into the open rather than paper over. + let remaining = match observer_regs + .release(&mut instances, &db_key, webview.label()) + .await + { + Some(remaining) => remaining, + None => { + // This webview was never registered as an observer of db_key - + // nothing to release, and nothing to tear down. + return Ok(()); + } + }; + if remaining > 0 { + debug!( + "unobserve: {} observer(s) remain for database {}, leaving broker active", + remaining, db_key + ); + return Ok(()); + } + + // Last observer released - fully tear down: abort subscriptions and disable + // the crate-level observer/broker. This `get_mut()` is unreachable-by + // -construction: `instances` was never dropped since the `contains_key` + // check above, so nothing could have removed the entry in between. The `?` + // is still here rather than `.expect()`/`.unwrap()` purely as a defensive + // fallback in case a future refactor breaks that invariant - it should + // never actually fire. + active_subs.remove_for_db(&db_key).await; let wrapper = instances .get_mut(&db_key) .ok_or_else(|| Error::DatabaseNotLoaded(db_key.clone()))?; - wrapper.disable_observation(); Ok(()) } diff --git a/src/error.rs b/src/error.rs index 6aaaa4f..ec8e3f7 100644 --- a/src/error.rs +++ b/src/error.rs @@ -55,6 +55,11 @@ pub enum Error { #[error("invalid configuration: {0}")] InvalidConfig(String), + /// A later `observe()` call requested a `channelCapacity`/`captureValues` + /// value that conflicts with the value already active for this database. + #[error("{0}")] + ObservationConfigConflict(String), + /// Required plugin managed state was not found. #[error("required plugin state not found: {0}")] MissingState(String), @@ -98,6 +103,7 @@ impl Error { Error::TooManyDatabases(_) => "TOO_MANY_DATABASES".to_string(), Error::TooManySubscriptions(_) => "TOO_MANY_SUBSCRIPTIONS".to_string(), Error::InvalidConfig(_) => "INVALID_CONFIG".to_string(), + Error::ObservationConfigConflict(_) => "OBSERVATION_CONFIG_CONFLICT".to_string(), Error::MissingState(_) => "MISSING_STATE".to_string(), Error::Other(_) => "ERROR".to_string(), } diff --git a/src/lib.rs b/src/lib.rs index 255a189..5529624 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,7 +24,7 @@ pub use sqlx_sqlite_toolkit::{ TransactionExecutionBuilder, WriteQueryResult, }; -use crate::subscriptions::ActiveSubscriptions; +use crate::subscriptions::{ActiveSubscriptions, ObserverRegistrations}; /// Default maximum number of concurrently loaded databases. const DEFAULT_MAX_DATABASES: usize = 50; @@ -533,6 +533,7 @@ impl Builder { }); app.manage(ActiveRegularTransactions::default()); app.manage(subscriptions::ActiveSubscriptions::default()); + app.manage(subscriptions::ObserverRegistrations::default()); // Run the deferred setup hook (if any), merge with static registrations. // Paths are validated and canonicalized at registration time. Hook errors @@ -627,6 +628,7 @@ impl Builder { let interruptible_txs_clone = app.state::().inner().clone(); let regular_txs_clone = app.state::().inner().clone(); let active_subs_clone = app.state::().inner().clone(); + let observer_regs_clone = app.state::().inner().clone(); // Run cleanup on the async runtime (without blocking the event loop), // then trigger a programmatic exit when done. ExitGuard ensures @@ -652,8 +654,10 @@ impl Builder { warn!("Transaction cleanup failed during exit: {e}"); } + // observer_regs_clone is cleared inside close_all_wrappers, + // under the same db lock used to drain the instances map. if let Err(e) = - close_all_wrappers(&instances_clone).await + close_all_wrappers(&instances_clone, &observer_regs_clone).await { warn!("Error closing databases during exit: {e:?}"); } @@ -689,6 +693,59 @@ impl Builder { } } } + // A window closing without ever calling unobserve() would otherwise + // leak its observer registration(s) forever, keeping affected + // brokers alive with no one left listening (the "phantom + // registration" gap from the #54 review). The webview label + // matches the window label for a `WebviewWindow` (the shape this + // whole design targets); a window hosting multiple independent + // webviews with distinct labels would need each webview's own + // Destroyed/close event, which this single window-level hook does + // not cover. Labels are also reusable across a window's lifetime: + // a window later recreated with the same static label silently + // "inherits" any registration left behind here - harmless for a + // fixed label, but worth knowing for dynamically labeled windows. + RunEvent::WindowEvent { + label, + event: tauri::WindowEvent::Destroyed, + .. + } => { + let observer_regs = app.state::().inner().clone(); + let active_subs = app.state::().inner().clone(); + let db_instances = app.state::().inner().clone(); + let label = label.clone(); + + tauri::async_runtime::spawn(async move { + // Lock order: db_instances write lock, then + // observer_regs (same order as observe()/unobserve() + // and close_database_inner - see the module doc in + // src/commands.rs). Held across the whole + // release+disable sequence: a concurrent observe() must + // not be able to register into a broker in the window + // between "registrations released" and "broker actually + // disabled" below. + let mut instances = db_instances.write().await; + let newly_unobserved = observer_regs + .release_all_for_label(&mut instances, &label) + .await; + if newly_unobserved.is_empty() { + return; + } + + debug!( + "Window '{}' destroyed - tearing down observation for {} database(s) with no remaining observers", + label, + newly_unobserved.len() + ); + + for db_key in newly_unobserved { + active_subs.remove_for_db(&db_key).await; + if let Some(wrapper) = instances.get_mut(&db_key) { + wrapper.disable_observation(); + } + } + }); + } _ => { // Other events don't require action } @@ -938,6 +995,9 @@ impl Connection for AppHandle { let subs = self .try_state::() .ok_or(Error::MissingState("ActiveSubscriptions".into()))?; + let observer_regs = self + .try_state::() + .ok_or(Error::MissingState("ObserverRegistrations".into()))?; let interruptible_txs = self .try_state::() @@ -951,6 +1011,7 @@ impl Connection for AppHandle { database_key, &instances, &subs, + &observer_regs, &interruptible_txs, ®ular_txs, ) @@ -1065,6 +1126,7 @@ pub(crate) async fn close_database( db_key: &str, db_instances: &DbInstances, active_subs: &ActiveSubscriptions, + observer_regs: &ObserverRegistrations, interruptible_txs: &ActiveInterruptibleTransactions, regular_txs: &ActiveRegularTransactions, ) -> Result { @@ -1075,6 +1137,7 @@ pub(crate) async fn close_database( &db_key, db_instances, active_subs, + observer_regs, interruptible_txs, regular_txs, ), @@ -1095,6 +1158,7 @@ async fn close_database_inner( db_key: &str, db_instances: &DbInstances, active_subs: &ActiveSubscriptions, + observer_regs: &ObserverRegistrations, interruptible_txs: &ActiveInterruptibleTransactions, regular_txs: &ActiveRegularTransactions, ) -> Result { @@ -1108,8 +1172,21 @@ async fn close_database_inner( last_error = Some(err.into()); } - let mut instances = db_instances.inner.write().await; + // Lock order: db_instances write lock, then observer_regs - same order as + // observe()/unobserve() (see the module doc in src/commands.rs), and for the + // same reason: clearing registrations while STILL holding the db lock (not + // before, and not after dropping it) prevents a concurrent observe() from + // registering into - and enabling observation on - this wrapper in the + // window between "registrations cleared" and "wrapper actually removed + // below". Without this, that race would leave a phantom registration for a + // wrapper that's about to be destroyed, i.e. reintroducing the exact + // problem this whole feature exists to prevent, for the close() path. + let mut instances = db_instances.write().await; let wrapper = instances.remove(db_key); + // Observation is torn down unconditionally on a full close, regardless of how + // many windows had registered via observe() - a closed database has no live + // broker for anyone to observe. + observer_regs.clear_for_db(&mut instances, db_key).await; drop(instances); let was_loaded = wrapper.is_some(); @@ -1125,9 +1202,17 @@ async fn close_database_inner( } } -async fn close_all_wrappers(db_instances: &DbInstances) -> Result<()> { - let mut instances = db_instances.inner.write().await; +/// Drains and closes every loaded database, clearing their observer +/// registrations under the same db lock (see `close_database_inner` for why +/// `observer_regs` must be touched while still holding `db_instances`'s lock, +/// not before or after). +async fn close_all_wrappers( + db_instances: &DbInstances, + observer_regs: &ObserverRegistrations, +) -> Result<()> { + let mut instances = db_instances.write().await; let wrappers: Vec = instances.drain().map(|(_, v)| v).collect(); + observer_regs.clear_all(&mut instances).await; drop(instances); let mut last_error: Option = None; @@ -1148,12 +1233,19 @@ async fn close_all_wrappers(db_instances: &DbInstances) -> Result<()> { pub(crate) async fn close_all_loaded_databases( db_instances: &DbInstances, active_subs: &ActiveSubscriptions, + observer_regs: &ObserverRegistrations, interruptible_txs: &ActiveInterruptibleTransactions, regular_txs: &ActiveRegularTransactions, ) -> Result<()> { let close_result = tokio::time::timeout( CLOSE_TIMEOUT, - close_all_loaded_databases_inner(db_instances, active_subs, interruptible_txs, regular_txs), + close_all_loaded_databases_inner( + db_instances, + active_subs, + observer_regs, + interruptible_txs, + regular_txs, + ), ) .await; @@ -1169,6 +1261,7 @@ pub(crate) async fn close_all_loaded_databases( async fn close_all_loaded_databases_inner( db_instances: &DbInstances, active_subs: &ActiveSubscriptions, + observer_regs: &ObserverRegistrations, interruptible_txs: &ActiveInterruptibleTransactions, regular_txs: &ActiveRegularTransactions, ) -> Result<()> { @@ -1182,7 +1275,9 @@ async fn close_all_loaded_databases_inner( last_error = Some(err.into()); } - if let Err(err) = close_all_wrappers(db_instances).await { + // observer_regs is cleared inside close_all_wrappers, under the same db + // lock used to drain the instances map - see its doc comment. + if let Err(err) = close_all_wrappers(db_instances, observer_regs).await { last_error = Some(err); } @@ -1212,6 +1307,7 @@ fn resolve_database_path(db_key: &str, app: &AppHandle) -> Result mod tests { use super::*; use crate::commands; + use crate::subscriptions::ObserverConfigParams; use std::collections::HashMap; use tauri::plugin::Plugin; use tauri::test::{MockRuntime, mock_app, mock_builder, mock_context, noop_assets}; @@ -1698,6 +1794,7 @@ mod tests { let closed = commands::close( app.state::(), app.state::(), + app.state::(), app.state::(), app.state::(), "MAIN".to_string(), @@ -1755,6 +1852,7 @@ mod tests { let closed = commands::close( app.state::(), app.state::(), + app.state::(), app.state::(), app.state::(), "MAIN".to_string(), @@ -1839,6 +1937,7 @@ mod tests { commands::close_all( app.state::(), app.state::(), + app.state::(), app.state::(), app.state::(), ) @@ -1919,6 +2018,7 @@ mod tests { let closed = commands::close( app.state::(), app.state::(), + app.state::(), app.state::(), app.state::(), "MAIN".to_string(), @@ -1969,6 +2069,121 @@ mod tests { }); } + /// Regression test for the app-wide freeze fixed alongside `remove()` + /// holding `db_instances`'s write lock across `wrapper.remove()`: an + /// abandoned interruptible transaction on the database being removed must + /// not stall operations on a *different*, unrelated loaded database, and + /// `remove()` itself must not hang forever waiting on a connection it is + /// itself holding. + /// + /// Before the fix, `wrapper.remove()`'s `Pool::close()` (no timeout of its + /// own) waited forever for the write connection an abandoned interruptible + /// transaction had checked out - `ActiveInterruptibleTransactions` only + /// reaps an abandoned transaction lazily, so nothing else was ever going + /// to release it. Because `db_instances`'s write lock is a single lock + /// shared by every loaded database, not one per key, that indefinite wait + /// blocked *every* database, not just the one being removed. + /// + /// Both assertions below are bounded well under `CLOSE_TIMEOUT` (5s) so + /// this test fails fast, rather than hanging the test binary, if either + /// half of the fix (transaction cleanup before teardown, or the + /// `CLOSE_TIMEOUT` wrap around the whole operation) regresses. + /// + /// What this does *not* prove: with the fix in place, `remove()`'s + /// critical section is fast enough (cleanup happens before the db lock is + /// even taken) that this test cannot reliably force the OTHER-database + /// query to land *inside* the brief window `remove()` still holds the + /// lock. That's fine for detecting a regression - in a reverted build the + /// lock is held for seconds, not microseconds, so the bounded query below + /// would time out regardless of exact scheduling - but it means a passing + /// run here doesn't demonstrate true concurrent interleaving on the fixed + /// code path, only that neither operation is ever left waiting past its + /// bound. + #[test] + fn test_remove_with_abandoned_transaction_does_not_stall_other_database() { + let temp_dir = tempfile::tempdir().unwrap(); + let main_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let other_path = validate::validate_database_path(temp_dir.path().join("other.db")).unwrap(); + + tauri::async_runtime::block_on(async { + let app = tokio::task::spawn_blocking(move || { + init_app_with_main_and_other(main_path, other_path) + }) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + load_and_create_test_table(&app, "OTHER").await; + + // Begin an interruptible transaction on MAIN and never continue, + // commit, or roll it back. This checks the write connection out of + // MAIN's pool for good unless something reclaims it - here, + // `remove()`'s own cleanup. + commands::begin_interruptible_transaction( + app.state::(), + app.state::(), + "MAIN".to_string(), + vec![Statement { + query: "INSERT INTO test (val) VALUES (?)".to_string(), + values: vec![serde_json::json!("abandoned")], + }], + None, + ) + .await + .expect("begin MAIN interruptible transaction should succeed"); + + let app_for_remove = app.handle().clone(); + let remove_task = tokio::spawn(async move { + commands::remove( + app_for_remove.state::(), + app_for_remove.state::(), + app_for_remove.state::(), + app_for_remove.state::(), + app_for_remove.state::(), + "MAIN".to_string(), + ) + .await + }); + + // Give the spawned remove() a chance to be scheduled and start + // running before we race it with the OTHER-database query below - + // mirrors the same dance in + // `test_execute_transaction_returns_cancelled_when_database_closed`. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let other_result = tokio::time::timeout( + std::time::Duration::from_secs(2), + commands::execute( + app.state::(), + "OTHER".to_string(), + "INSERT INTO test (val) VALUES ('unaffected')".to_string(), + vec![], + None, + ), + ) + .await; + + assert!( + other_result.is_ok(), + "a query on an unrelated database must not be blocked by remove() cleaning up \ + an abandoned transaction on a different database" + ); + other_result + .unwrap() + .expect("OTHER query should succeed while MAIN is being removed"); + + let removed = tokio::time::timeout(std::time::Duration::from_secs(2), remove_task) + .await + .expect( + "remove() must not hang forever on an abandoned transaction - it should finish \ + well within its own CLOSE_TIMEOUT", + ) + .expect("remove task should not panic") + .expect("remove() should succeed once the abandoned transaction is cleaned up"); + assert!(removed, "MAIN was loaded and should have been removed"); + }); + } + #[test] fn test_execute_transaction_returns_cancelled_when_database_closed() { let temp_dir = tempfile::tempdir().unwrap(); @@ -2008,6 +2223,7 @@ mod tests { commands::close( app.state::(), app.state::(), + app.state::(), app.state::(), app.state::(), "MAIN".to_string(), @@ -2023,4 +2239,1380 @@ mod tests { assert!(err.to_string().contains("transaction cancelled")); }); } + + /// Returns whether observation is currently enabled for `db_key`. + async fn is_observing(app: &tauri::App, db_key: &str) -> bool { + app.state::() + .inner() + .inner + .read() + .await + .get(db_key) + .expect("database should be loaded") + .is_observing() + } + + /// Polls `db_instances`'s write lock via `try_write()` until some other + /// task is holding it (i.e. our own `try_write()` starts failing), instead + /// of assuming a fixed sleep gave a spawned task enough time to reach the + /// point of contention. + /// + /// Used by the deterministic lock-order tests below: without this, a fixed + /// sleep followed by a single timed acquisition attempt is a CI flake risk. + /// If task scheduling is slow enough that the spawned command hasn't taken + /// the db lock yet by the time the sleep ends, the acquisition attempt + /// succeeds (there's nothing contending it yet) and the test fails for a + /// reason that has nothing to do with the invariant it's guarding. + /// Polling for the *first sign* of contention, with a + /// generous overall budget, turns "was the task scheduled fast enough" + /// from a pass/fail race into something we simply wait out. + async fn wait_until_db_instances_write_contended( + app: &tauri::App, + budget: std::time::Duration, + ) { + let deadline = std::time::Instant::now() + budget; + loop { + if app + .state::() + .inner() + .inner + .try_write() + .is_err() + { + return; + } + assert!( + std::time::Instant::now() < deadline, + "db_instances write lock was never observed to be contended within {budget:?} - \ + the command under test may not have started, or may not be taking the lock at all" + ); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + } + + /// Regression test for issue #54: re-calling `observe()` (from the same or a + /// different window) must not tear down subscriptions that were already active. + /// A subscriber created before a second `observe()` call - with a *different* + /// table set - must still receive events published after that second call. + #[test] + fn test_observe_reenable_with_different_tables_preserves_existing_subscriber() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + commands::execute( + app.state::(), + "MAIN".to_string(), + "CREATE TABLE other (id INTEGER PRIMARY KEY, val TEXT)".to_string(), + vec![], + None, + ) + .await + .expect("create other table should succeed"); + + let webview = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window should build"); + + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("first observe should succeed"); + + let (tx, rx) = std::sync::mpsc::channel::(); + let channel = tauri::ipc::Channel::new(move |body| { + let value: serde_json::Value = body + .deserialize() + .expect("payload should deserialize as JSON"); + tx.send(value).ok(); + Ok(()) + }); + + commands::subscribe( + app.state::(), + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + channel, + ) + .await + .expect("subscribe should succeed"); + + // Re-observe with a DIFFERENT table set. Under the old (destructive) + // behavior, this would drop the broker and silently end the + // subscription created above. + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["other".to_string()], + None, + ) + .await + .expect("second observe should succeed"); + + commands::execute( + app.state::(), + "MAIN".to_string(), + "INSERT INTO test (val) VALUES ('hello')".to_string(), + vec![], + None, + ) + .await + .expect("insert should succeed"); + + let received = tokio::task::spawn_blocking(move || { + rx.recv_timeout(std::time::Duration::from_millis(500)) + }) + .await + .expect("blocking recv task should not panic"); + + let payload = + received.expect("subscriber should still receive events after the second observe()"); + assert_eq!(payload["event"], "change"); + assert_eq!(payload["data"]["table"], "test"); + }); + } + + /// Refcount teardown boundary: the broker stays live while at least one + /// window is registered as an observer, and is only torn down once the last + /// registered window releases via `unobserve()`. A non-final `unobserve()` + /// must leave every other window's subscriptions running untouched (#54); + /// only the final `unobserve()` - the one that drops the refcount to zero - + /// aborts them. + #[test] + fn test_unobserve_refcount_teardown_boundary() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window A observe should succeed"); + + commands::observe( + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window B observe should succeed"); + + assert!( + is_observing(&app, "MAIN").await, + "broker should be live with two registered observers" + ); + + // Window A subscribes before releasing its observer registration, so + // window A's own (non-final) unobserve() below can be checked for not + // aborting a subscription that isn't its own to tear down. + let (tx, rx) = std::sync::mpsc::channel::(); + let channel = tauri::ipc::Channel::new(move |body| { + let value: serde_json::Value = body + .deserialize() + .expect("payload should deserialize as JSON"); + tx.send(value).ok(); + Ok(()) + }); + + commands::subscribe( + app.state::(), + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + channel, + ) + .await + .expect("subscribe on window A should succeed"); + + commands::unobserve( + app.state::(), + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + ) + .await + .expect("window A unobserve should succeed"); + + assert!( + is_observing(&app, "MAIN").await, + "broker should stay live while window B is still registered (rc=1)" + ); + + // Timing-free: catches a non-final unobserve() aborting subscriptions + // it doesn't own (moving/duplicating the `remove_for_db` call above + // the `remaining > 0` early return - see #54). + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 1, + "a non-final unobserve() must not abort another window's subscription (#54)" + ); + + // Real round-trip: the only thing proving events still actually flow + // to the surviving subscription after window A's unobserve(). + commands::execute( + app.state::(), + "MAIN".to_string(), + "INSERT INTO test (val) VALUES ('hello')".to_string(), + vec![], + None, + ) + .await + .expect("insert should succeed"); + + let received = + tokio::task::spawn_blocking(move || rx.recv_timeout(std::time::Duration::from_secs(2))) + .await + .expect("blocking recv task should not panic"); + + let payload = received + .expect("subscription should still receive events after a non-final unobserve()"); + assert_eq!(payload["event"], "change"); + assert_eq!(payload["data"]["table"], "test"); + + commands::unobserve( + app.state::(), + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + ) + .await + .expect("window B unobserve should succeed"); + + assert!( + !is_observing(&app, "MAIN").await, + "broker should be torn down once the last observer releases (rc=0)" + ); + + // Catches deleting the `remove_for_db` call entirely: without it, + // nothing asserts the final unobserve() aborts subscriptions. Not + // deterministic, though: subscribe()'s own forwarding task self-reaps + // its `active_subs` entry once its stream ends (see the reap comment + // in `commands::subscribe`), and tearing down the broker here also + // ends that stream - so the self-reaper can independently drive this + // count to 0 even with `remove_for_db` deleted, racing the assertion + // below. Measured detection of that mutation: 9 of 10 runs. + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 0, + "the last unobserve() must abort remaining subscriptions" + ); + }); + } + + /// A second window's explicit `captureValues` request that conflicts with + /// the value already active for a database must be rejected with + /// `OBSERVATION_CONFIG_CONFLICT`, without mutating the live broker or + /// recording that window's registration - the check must return before + /// `register()` runs. + #[test] + fn test_observe_conflicting_capture_values_rejected() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window A observe with default config should succeed"); + + let err = commands::observe( + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + Some(ObserverConfigParams { + channel_capacity: None, + capture_values: Some(false), + }), + ) + .await + .expect_err("conflicting captureValues should be rejected"); + + assert!(matches!(err, Error::ObservationConfigConflict(_))); + + let (capacity, capture) = { + let instances = app.state::().inner().inner.read().await; + let observable = instances + .get("MAIN") + .expect("MAIN should be loaded") + .observable() + .expect("observation should be enabled"); + ( + observable.broker().channel_capacity(), + observable.broker().capture_values(), + ) + }; + assert_eq!( + capacity, 256, + "the live broker's channelCapacity must be unchanged by a rejected observe()" + ); + assert!( + capture, + "the live broker's captureValues must be unchanged by a rejected observe()" + ); + + assert!( + !app + .state::() + .is_registered("MAIN", webview_b.label()) + .await, + "window B's registration must not be recorded when its observe() is rejected" + ); + }); + } + + /// Same as `test_observe_conflicting_capture_values_rejected`, but for a + /// conflicting `channelCapacity` request. + #[test] + fn test_observe_conflicting_channel_capacity_rejected() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window A observe with default config should succeed"); + + let err = commands::observe( + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + Some(ObserverConfigParams { + channel_capacity: Some(512), + capture_values: None, + }), + ) + .await + .expect_err("conflicting channelCapacity should be rejected"); + + assert!(matches!(err, Error::ObservationConfigConflict(_))); + + let (capacity, capture) = { + let instances = app.state::().inner().inner.read().await; + let observable = instances + .get("MAIN") + .expect("MAIN should be loaded") + .observable() + .expect("observation should be enabled"); + ( + observable.broker().channel_capacity(), + observable.broker().capture_values(), + ) + }; + assert_eq!( + capacity, 256, + "the live broker's channelCapacity must be unchanged by a rejected observe()" + ); + assert!( + capture, + "the live broker's captureValues must be unchanged by a rejected observe()" + ); + + assert!( + !app + .state::() + .is_registered("MAIN", webview_b.label()) + .await, + "window B's registration must not be recorded when its observe() is rejected" + ); + }); + } + + /// An explicit config that matches the live broker's already-active values + /// exactly is not a conflict and must succeed. + #[test] + fn test_observe_identical_explicit_config_succeeds() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window A observe with default config should succeed"); + + commands::observe( + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + Some(ObserverConfigParams { + channel_capacity: Some(256), + capture_values: Some(true), + }), + ) + .await + .expect("explicit config identical to the live broker's values should succeed"); + + assert!( + app.state::() + .is_registered("MAIN", webview_b.label()) + .await, + "window B's registration must be recorded once its observe() succeeds" + ); + }); + } + + /// Guards the `seeded` block in `observe()`: once window A has enabled + /// observation with a non-default explicit config, a second window calling + /// `observe()` with `config: None` must succeed - its defaults must never be + /// compared against the live broker's (already non-default) values as if + /// they were an explicit, conflicting request. + #[test] + fn test_observe_omitted_config_after_explicit_config_succeeds() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + Some(ObserverConfigParams { + channel_capacity: Some(512), + capture_values: Some(false), + }), + ) + .await + .expect("window A observe with explicit non-default config should succeed"); + + commands::observe( + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window B observe with config: None should succeed"); + + assert!( + app.state::() + .is_registered("MAIN", webview_b.label()) + .await, + "window B's registration must be recorded once its observe() succeeds" + ); + }); + } + + /// `subscribe()` must reject a window that never called `observe()` for + /// `db_key` itself with `OBSERVATION_NOT_ENABLED`, even while another + /// window's registration keeps a broker active for that database (#54), + /// and must not disturb that other window's already-established + /// subscription. + #[test] + fn test_subscribe_without_observe_rejected() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("window A observe should succeed"); + + let (tx, rx) = std::sync::mpsc::channel::(); + let channel = tauri::ipc::Channel::new(move |body| { + let value: serde_json::Value = body + .deserialize() + .expect("payload should deserialize as JSON"); + tx.send(value).ok(); + Ok(()) + }); + + commands::subscribe( + app.state::(), + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + channel, + ) + .await + .expect("subscribe on window A, which observed, should succeed"); + + // Window B never called observe() - the broker is only live because + // of window A's registration. + let (tx_b, _rx_b) = std::sync::mpsc::channel::(); + let channel_b = tauri::ipc::Channel::new(move |body| { + let value: serde_json::Value = body + .deserialize() + .expect("payload should deserialize as JSON"); + tx_b.send(value).ok(); + Ok(()) + }); + + let err = commands::subscribe( + app.state::(), + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + channel_b, + ) + .await + .expect_err("subscribe from a window that never observed should be rejected"); + + assert!(matches!(err, Error::ObservationNotEnabled(_))); + + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 1, + "window A's subscription must be unaffected by window B's rejected subscribe()" + ); + + commands::execute( + app.state::(), + "MAIN".to_string(), + "INSERT INTO test (val) VALUES ('hello')".to_string(), + vec![], + None, + ) + .await + .expect("insert should succeed"); + + let received = + tokio::task::spawn_blocking(move || rx.recv_timeout(std::time::Duration::from_secs(2))) + .await + .expect("blocking recv task should not panic"); + + let payload = received + .expect("window A's subscription should still receive events after window B's rejected subscribe()"); + assert_eq!(payload["event"], "change"); + assert_eq!(payload["data"]["table"], "test"); + }); + } + + /// Lock-order regression guard: `observe()`/`unobserve()` must hold a single lock + /// (db_instances, then observer_regs) across their whole enable/register or + /// release/disable sequence, or a concurrent pair on different webviews can + /// interleave such that `is_observing()` and "has registrations" disagree - + /// e.g. a window's `observe()` registers into a broker a concurrent + /// `unobserve()` from another window just destroyed. + /// + /// This is a probabilistic guard, not a proof, and its detection rate is + /// asymmetric: a regressed `unobserve()` side fails reliably, but a regressed + /// `observe()` side has been measured as low as 1 in 10 runs. `ITERATIONS` is + /// 2000 because more attempts are the only lever available against a window + /// this narrow, not because a higher count is known to detect more - that has + /// not been demonstrated. At ~0.14s the attempts are cheap either way. + /// Re-measure before lowering it, and do not rely on this test alone. + /// + /// Do not delete this test as redundant. It is the *only* guard for a + /// distinct regression the other two structurally cannot see: releasing the + /// db guard mid-sequence and immediately reacquiring a fresh one before + /// `register()`/`release()`. That shape compiles (a real guard satisfies the + /// `DbInstancesGuard` witness) and passes both deterministic tests 3/3 (a + /// guard genuinely *is* held at the call), yet still reopens the race - a + /// concurrent `unobserve()` can take the db lock in the drop/reacquire + /// window, see the refcount reach zero, and tear the broker down before + /// `observe()` reacquires and registers. Mutation testing measured this + /// test failing 5/5 against that shape while both deterministic guards + /// passed 3/3. + #[test] + fn test_concurrent_observe_and_unobserve_keep_broker_and_registrations_in_sync() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + const ITERATIONS: usize = 2000; + + for i in 0..ITERATIONS { + // Baseline for this iteration: only window A registered, broker + // live. Force both released first (no-ops if already released) + // so each iteration starts from a known state regardless of how + // the previous iteration's race resolved. + let _ = commands::unobserve( + app.state::(), + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + ) + .await; + let _ = commands::unobserve( + app.state::(), + app.state::(), + app.state::(), + webview_b.as_ref().clone(), + "MAIN".to_string(), + ) + .await; + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("baseline observe for window A should succeed"); + + // Race: window B observes while window A unobserves, concurrently, + // on separate spawned tasks so the multi-thread runtime can + // actually run them in parallel rather than just interleaving at + // await points on one thread. + let app_for_observe = app.handle().clone(); + let webview_b_for_task = webview_b.as_ref().clone(); + let observe_task = tokio::spawn(async move { + commands::observe( + app_for_observe.state::(), + app_for_observe.state::(), + webview_b_for_task, + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + }); + + let app_for_unobserve = app.handle().clone(); + let webview_a_for_task = webview_a.as_ref().clone(); + let unobserve_task = tokio::spawn(async move { + commands::unobserve( + app_for_unobserve.state::(), + app_for_unobserve.state::(), + app_for_unobserve.state::(), + webview_a_for_task, + "MAIN".to_string(), + ) + .await + }); + + observe_task + .await + .expect("observe task should not panic") + .expect("observe should succeed"); + unobserve_task + .await + .expect("unobserve task should not panic") + .expect("unobserve should succeed"); + + let has_registrations = app + .state::() + .count_for_db("MAIN") + .await + > 0; + let observing = is_observing(&app, "MAIN").await; + + assert_eq!( + observing, has_registrations, + "iteration {i}: is_observing() ({observing}) and has-registrations \ + ({has_registrations}) must always agree" + ); + } + }); + } + + /// Deterministic lock-order guard, observe side. + /// + /// Forces the exact contention the lock-order invariant depends on instead + /// of relying on scheduling luck: the test itself holds `observer_regs`'s + /// lock (via the test-only `lock_for_test()` accessor), so `observe()`'s + /// `register()` call is guaranteed to block. If `observe()` is correctly + /// holding the `db_instances` lock across that whole sequence, its lock + /// guard is still alive while blocked - so this test's own attempt to + /// acquire that same lock (with a short timeout) MUST fail. A regressed + /// `observe()` that drops the db lock before calling `register()` would let + /// this acquisition succeed instead. + #[test] + fn test_observe_holds_db_lock_across_register() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window should build"); + + // Hold observer_regs' lock ourselves so observe()'s register() call + // is guaranteed to block on it. + let observer_regs_state = app.state::(); + let regs_guard = observer_regs_state.lock_for_test().await; + + let app_for_observe = app.handle().clone(); + let webview_for_task = webview.as_ref().clone(); + let observe_task = tokio::spawn(async move { + commands::observe( + app_for_observe.state::(), + app_for_observe.state::(), + webview_for_task, + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + }); + + // Wait for the spawned task to actually reach the point of + // contention (acquire the db lock, call enable_observation, reach + // register(), and block on the regs lock we're holding), rather than + // assuming a fixed sleep was long enough. + wait_until_db_instances_write_contended(&app, std::time::Duration::from_secs(2)).await; + + let db_lock_attempt = tokio::time::timeout( + std::time::Duration::from_millis(200), + app.state::().inner().inner.write(), + ) + .await; + + assert!( + db_lock_attempt.is_err(), + "observe() must still be holding the db_instances lock while blocked on \ + observer_regs.register() - if this acquisition succeeded, observe() had \ + already dropped the db lock before registering (the exact regression this \ + test guards against)" + ); + + // Release the regs lock so observe() can finish, then clean up. + drop(regs_guard); + + observe_task + .await + .expect("observe task should not panic") + .expect("observe should succeed"); + }); + } + + /// Deterministic lock-order guard, unobserve side. + /// + /// Same technique as `test_observe_holds_db_lock_across_register`, but for + /// `unobserve()`'s db-lock-then-release() ordering. + #[test] + fn test_unobserve_holds_db_lock_across_release() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window should build"); + + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("baseline observe should succeed"); + + // Hold observer_regs' lock ourselves so unobserve()'s release() call + // is guaranteed to block on it. + let observer_regs_state = app.state::(); + let regs_guard = observer_regs_state.lock_for_test().await; + + let app_for_unobserve = app.handle().clone(); + let webview_for_task = webview.as_ref().clone(); + let unobserve_task = tokio::spawn(async move { + commands::unobserve( + app_for_unobserve.state::(), + app_for_unobserve.state::(), + app_for_unobserve.state::(), + webview_for_task, + "MAIN".to_string(), + ) + .await + }); + + wait_until_db_instances_write_contended(&app, std::time::Duration::from_secs(2)).await; + + let db_lock_attempt = tokio::time::timeout( + std::time::Duration::from_millis(200), + app.state::().inner().inner.write(), + ) + .await; + + assert!( + db_lock_attempt.is_err(), + "unobserve() must still be holding the db_instances lock while blocked on \ + observer_regs.release() - if this acquisition succeeded, unobserve() had \ + already released the db lock before calling release() (the exact \ + regression this test guards against)" + ); + + drop(regs_guard); + + unobserve_task + .await + .expect("unobserve task should not panic") + .expect("unobserve should succeed"); + }); + } + + /// Closing a database fully clears its observer registrations (via + /// `ObserverRegistrations::clear_for_db`, wired through + /// `close_database_inner`), so a fresh `observe()` after a `close()`+reload + /// cycle restarts the refcount at 1 rather than inheriting stale entries + /// from before the close. + #[test] + fn test_observe_after_close_restarts_refcount() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window should build"); + + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("observe should succeed"); + + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 1 + ); + + commands::close( + app.state::(), + app.state::(), + app.state::(), + app.state::(), + app.state::(), + "MAIN".to_string(), + ) + .await + .expect("close should succeed"); + + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 0, + "close() should clear stale observer registrations, not just the crate-level broker" + ); + + // Reload and recreate the table (close() dropped the connection pool; + // the underlying file-backed table itself still exists on disk, but + // a fresh wrapper needs to be loaded before observe() will find it). + connect_to_database(app.handle(), "MAIN", None) + .await + .expect("reconnect after close should succeed"); + + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("observe after reload should succeed"); + + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 1, + "refcount should restart at 1, not inherit anything from before the close" + ); + assert!(is_observing(&app, "MAIN").await); + }); + } + + /// A window calling `observe()` twice (e.g. to add more tables) still only + /// holds ONE registration, so a single `unobserve()` call from that same + /// window fully tears the broker down - the refcount + /// tracks distinct webviews, not the number of `observe()` calls made. + #[test] + fn test_same_window_double_observe_then_single_unobserve_tears_down() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + commands::execute( + app.state::(), + "MAIN".to_string(), + "CREATE TABLE other (id INTEGER PRIMARY KEY, val TEXT)".to_string(), + vec![], + None, + ) + .await + .expect("create other table should succeed"); + + let webview = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window should build"); + + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("first observe should succeed"); + + // Same window, second call, different tables - must not inflate the + // refcount for this window. + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["other".to_string()], + None, + ) + .await + .expect("second observe should succeed"); + + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 1, + "same window calling observe() twice must still be a single registration" + ); + + commands::unobserve( + app.state::(), + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + ) + .await + .expect("unobserve should succeed"); + + assert!( + !is_observing(&app, "MAIN").await, + "a single unobserve() from the only registered window must fully tear down the broker" + ); + }); + } + + /// Probabilistic regression guard for the close()-vs-observe() variant of + /// the lock-order invariant: `close_database_inner` must clear `observer_regs` + /// while still holding the `db_instances` write lock used to remove the + /// wrapper (not before acquiring it), or a concurrent `observe()` from + /// another window can register into - and enable observation on - a + /// wrapper that's about to be removed, leaving a phantom registration + /// behind after `close()` completes with no wrapper left for it to refer + /// to. `close()` always removes the wrapper by the time it returns, so the + /// database is deterministically unloaded after each iteration; what's + /// racy is only whether a registration is left behind. + #[test] + fn test_concurrent_observe_and_close_do_not_leak_registrations() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview_a = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window A should build"); + let webview_b = tauri::WebviewWindowBuilder::new(&app, "window-b", Default::default()) + .build() + .expect("webview window B should build"); + + const ITERATIONS: usize = 200; + + for i in 0..ITERATIONS { + // Baseline for this iteration: MAIN loaded, window A observing. + connect_to_database(app.handle(), "MAIN", None) + .await + .expect("reconnect should succeed"); + commands::observe( + app.state::(), + app.state::(), + webview_a.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("baseline observe for window A should succeed"); + + // Race: window B observes while the database is closed, concurrently. + let app_for_observe = app.handle().clone(); + let webview_b_for_task = webview_b.as_ref().clone(); + let observe_task = tokio::spawn(async move { + commands::observe( + app_for_observe.state::(), + app_for_observe.state::(), + webview_b_for_task, + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + }); + + let app_for_close = app.handle().clone(); + let close_task = tokio::spawn(async move { + commands::close( + app_for_close.state::(), + app_for_close.state::(), + app_for_close.state::(), + app_for_close.state::(), + app_for_close.state::(), + "MAIN".to_string(), + ) + .await + }); + + // observe() may legitimately fail with DATABASE_NOT_LOADED if + // close() won the race for the db lock first - that's fine, not + // what this test is guarding against. + let _ = observe_task.await.expect("observe task should not panic"); + close_task + .await + .expect("close task should not panic") + .expect("close should succeed"); + + // close() always removes the wrapper by the time it returns, so + // the database is deterministically unloaded here - use is_some() + // directly rather than the is_observing() helper, which assumes a + // loaded database and would itself panic. + assert!( + app.state::() + .inner() + .inner + .read() + .await + .get("MAIN") + .is_none(), + "iteration {i}: close() should always leave the database unloaded" + ); + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 0, + "iteration {i}: no registration should survive close() with no wrapper left to refer to" + ); + } + }); + } + + /// Pins the invariant that once a subscription's forwarding task ends (its + /// channel closed), it reaps its own entry from `ActiveSubscriptions` - + /// exercising the `oneshot` ready-gate + self-removal logic rather than + /// only reasoning about it. The channel below models the event loop being + /// gone entirely (every `send` fails), not a same-webview reload - see the + /// reap comment in `commands::subscribe` for why a reload can't actually + /// trigger this path. Without this reap, letting the event loop/broker tear + /// down would leave a dead entry behind forever, and eventually every *new* + /// subscribe() call would hit `TOO_MANY_SUBSCRIPTIONS` even though nothing + /// is actually subscribed anymore. + #[test] + fn test_subscribe_reaps_itself_after_channel_closes() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = validate::validate_database_path(temp_dir.path().join("main.db")).unwrap(); + let key = "MAIN".to_string(); + + tauri::async_runtime::block_on(async { + let (app, _) = + tokio::task::spawn_blocking(move || init_app_with_registered_db_at_path(&key, db_path)) + .await + .expect("plugin init task should succeed"); + + load_and_create_test_table(&app, "MAIN").await; + + let webview = tauri::WebviewWindowBuilder::new(&app, "window-a", Default::default()) + .build() + .expect("webview window should build"); + + commands::observe( + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + None, + ) + .await + .expect("observe should succeed"); + + // A channel whose "send" always fails, simulating the event loop + // being gone entirely (not a reload/navigation - a live webview's + // `Channel::send` still reports `Ok` even when nothing is listening + // on the JS side; see the reap comment in `commands::subscribe`). The + // forwarding task's `on_event.send(...).is_err()` check will be true + // on the first change event, causing it to break out of its loop and + // reap itself. + let channel = tauri::ipc::Channel::new(|_body| { + Err(std::io::Error::other("simulated closed channel").into()) + }); + + commands::subscribe( + app.state::(), + app.state::(), + app.state::(), + webview.as_ref().clone(), + "MAIN".to_string(), + vec!["test".to_string()], + channel, + ) + .await + .expect("subscribe should succeed"); + + assert_eq!( + app.state::() + .count_for_db("MAIN") + .await, + 1, + "subscription should be registered immediately after subscribe()" + ); + + commands::execute( + app.state::(), + "MAIN".to_string(), + "INSERT INTO test (val) VALUES ('trigger')".to_string(), + vec![], + None, + ) + .await + .expect("insert should succeed"); + + // Give the forwarding task a chance to receive the change, fail to + // send it through the closed channel, and reap itself. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + if app + .state::() + .count_for_db("MAIN") + .await + == 0 + { + break; + } + assert!( + std::time::Instant::now() < deadline, + "forwarding task never reaped its own entry after its channel closed" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }); + } } diff --git a/src/subscriptions.rs b/src/subscriptions.rs index 21207a2..ec3b100 100644 --- a/src/subscriptions.rs +++ b/src/subscriptions.rs @@ -4,15 +4,72 @@ //! Tauri's IPC layer, converting observer types to serializable payloads and //! managing active subscription state. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use serde::{Deserialize, Serialize}; -use tokio::sync::RwLock; +use sqlx_sqlite_toolkit::DatabaseWrapper; +use tokio::sync::{RwLock, RwLockWriteGuard}; use tracing::debug; use sqlx_sqlite_observer::{ChangeOperation, ColumnValue, TableChange, TableChangeEvent}; +/// A held write guard on `DbInstances`'s inner map, passed to every mutating +/// `ObserverRegistrations` method as proof the caller already holds it. +/// +/// This is a **witness, not a resource**: these methods never read or write +/// through it, only require its existence for the duration of the call. See +/// `ObserverRegistrations`'s doc comment and the module doc in +/// `src/commands.rs` for why - do not "simplify" a call site by fetching a +/// value out of it. +/// +/// The inner guard is intentionally private: that makes +/// [`DbInstances::write`](crate::DbInstances::write) the only way to construct +/// one, so "no db guard held at all" - the shape that actually regressed four +/// times - is a compile error at every call site. Were the field public, or +/// were this a plain type alias for +/// `RwLockWriteGuard<'_, HashMap>`, *any* write guard +/// on *any* `HashMap` would satisfy the parameter. +/// +/// What the token does *not* prove is which `DbInstances` the guard came from. +/// `DbInstances` is `pub`, implements `Default`, and has a `pub fn new`, so +/// `DbInstances::default().write().await` yields a perfectly valid witness over +/// an empty throwaway map - this module's own tests rely on exactly that (see +/// `tests::dummy_db_lock`). So it means "the caller holds *a* `DbInstances` +/// write guard"; that it is *this app's* rests on Tauri managing one instance +/// per type, not on the type system. Nor can it see acquisition order, or a +/// guard dropped and reacquired mid-sequence - see the module doc in +/// `src/commands.rs` for which tests cover those, and how well. +pub struct DbInstancesGuard<'a>(RwLockWriteGuard<'a, HashMap>); + +impl std::ops::Deref for DbInstancesGuard<'_> { + type Target = HashMap; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl std::ops::DerefMut for DbInstancesGuard<'_> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 + } +} + +impl crate::DbInstances { + /// Acquires this `DbInstances`'s write lock, wrapped as the + /// [`DbInstancesGuard`] witness that every mutating `ObserverRegistrations` + /// method requires. This is the *only* way to construct that witness - see + /// its doc comment for what that does and does not prove. + /// + /// Defined here, alongside `DbInstancesGuard`, rather than next to + /// `DbInstances`'s own definition - that's what lets it (and only it) + /// reach `DbInstancesGuard`'s private field. + pub(crate) async fn write(&self) -> DbInstancesGuard<'_> { + DbInstancesGuard(self.inner.write().await) + } +} + /// Serializable column value for IPC transport. /// /// Maps observer's `ColumnValue` to a tagged enum that can be sent to the frontend. @@ -110,6 +167,19 @@ pub struct ObserverConfigParams { } /// Tracks an active subscription's abort handle. +/// +/// # These field types are load-bearing for lock safety +/// +/// `active_subs` is locked in both orders relative to `db_instances`'s lock +/// (see the "`ActiveSubscriptions` is a separate pair" section of the module +/// doc in `src/commands.rs`). That is deadlock-free only because nothing +/// reachable from this struct can reach back for another lock: an +/// `AbortHandle` schedules cancellation on the runtime rather than running +/// destructors inline, and a `String` does nothing at all. Adding a field +/// holding a `DatabaseWrapper`, a `DbInstances`, an `AppHandle`, or anything +/// whose `Drop` touches plugin state would let dropping an entry inside +/// `ActiveSubscriptions`'s own lock reach for the db lock - turning that +/// inconsistent order into a real lock cycle. struct ActiveSubscription { /// Abort handle for the subscription forwarding task. abort_handle: tokio::task::AbortHandle, @@ -176,3 +246,471 @@ impl ActiveSubscriptions { } } } + +/// Tracks which webview windows currently hold an active `observe()` registration +/// for each database, keyed by database key and webview label. +/// +/// Observation is additive and reference-counted: multiple windows can call +/// `observe()` on the same database independently, and the underlying broker +/// (and its subscribers) is only torn down once every window that registered has +/// released its registration via `unobserve()`. See issue #54 — previously, +/// re-calling `observe()` unconditionally destroyed the existing broker, silently +/// terminating every other window's subscriptions. +/// +/// The webview label is used as the observer identity because it is the only +/// caller-scoped handle already available to Tauri commands without adding a new +/// argument to the JS-facing API. +/// +/// # Lock order: `DbInstances` first +/// +/// Every mutating method here (`register`, `release`, `release_all_for_label`, +/// `clear_for_db`, `clear_all`) takes a [`DbInstancesGuard`] as its first +/// parameter - a witness proving the caller already holds `DbInstances`'s write +/// lock. +/// +/// The defect this prevents: mutating `observer_regs` without holding the +/// `db_instances` lock lets a concurrent `observe()` register into - or a +/// concurrent teardown destroy - a broker the other side doesn't know is being +/// touched, leaving the refcount and the broker's actual state disagreeing. +/// That shape was reintroduced four separate times while the rule existed only +/// as prose in this comment. The witness makes a call site holding *no* db +/// guard fail to compile instead of merely being wrong. It cannot see +/// acquisition order, or a guard dropped and reacquired mid-sequence; those are +/// covered by tests only, one of them probabilistically - see the module doc in +/// `src/commands.rs`. +#[derive(Clone, Default)] +pub struct ObserverRegistrations(Arc>>>); + +impl ObserverRegistrations { + /// Acquires and holds this registry's internal write lock. + /// + /// Test-only. Lets a test hold the `observer_regs` lock itself and then + /// assert that a concurrent `observe()`/`unobserve()` call blocks trying to + /// acquire it - which can only happen if that call is still holding the + /// `db_instances` lock at the point it reaches its `register()`/`release()` + /// call. This proves the lock-order invariant deterministically, rather than + /// relying on probabilistic scheduling to expose a violation. + #[cfg(test)] + pub async fn lock_for_test( + &self, + ) -> tokio::sync::RwLockWriteGuard<'_, HashMap>> { + self.0.write().await + } + + /// Registers `webview_label` as an observer of `db_key`. + /// + /// Idempotent: registering the same label for the same database more than once + /// (e.g. a window calling `observe()` again to add more tables) does not + /// increase the refcount. Returns the number of distinct observing webviews for + /// this database after registering. + /// + /// # Granularity is per webview, not per caller + /// + /// Because the identity is the webview label, that idempotency is not limited + /// to one logical caller. Two independent frontend modules in the *same* + /// window that each call `observe()` collapse into a single registration, so + /// whichever of them calls `unobserve()` first drives the refcount to zero and + /// triggers the full teardown - aborting the other module's subscriptions. A + /// window therefore needs a single owner of the `observe()`/`unobserve()` + /// pair; "reference-counted" does not make `unobserve()` locally safe within a + /// window. `subscribe()`'s registration check does not help here either, since + /// both modules share the label and both pass it. + /// + /// Note that scoping teardown by webview label would not fix this: both + /// modules share the label, so a label-scoped abort covers the identical set, + /// and teardown drops the broker regardless. Fixing it properly needs a + /// per-caller registration token or a frontend-side refcount, which is a + /// public API change. + /// + /// `_db_guard` is a witness that the caller already holds `DbInstances`'s + /// write lock for the duration of this call - see the lock-order doc above. + /// It is never read through. + pub async fn register( + &self, + _db_guard: &mut DbInstancesGuard<'_>, + db_key: &str, + webview_label: &str, + ) -> usize { + let mut regs = self.0.write().await; + let labels = regs.entry(db_key.to_string()).or_default(); + labels.insert(webview_label.to_string()); + labels.len() + } + + /// Releases `webview_label`'s observation registration for `db_key`. + /// + /// Returns `None` if `webview_label` was never registered as an observer of + /// `db_key`, which the caller must treat as "nothing to do" rather than + /// as "the last observer just left". Collapsing both to a plain `0` would + /// mean a window calling `unobserve()` without ever having called + /// `observe()` triggers a full broker teardown - destroying a broker other + /// windows are still legitimately using. + /// + /// Otherwise returns `Some(remaining)`, the number of distinct observing + /// webviews left registered for this database. `Some(0)` means this was the + /// last registered observer and the caller should tear down the broker (via + /// `DatabaseWrapper::disable_observation`) and any remaining subscriptions. + /// + /// `_db_guard` is a witness - see [`register`](Self::register). + pub async fn release( + &self, + _db_guard: &mut DbInstancesGuard<'_>, + db_key: &str, + webview_label: &str, + ) -> Option { + let mut regs = self.0.write().await; + let labels = regs.get_mut(db_key)?; + + if !labels.remove(webview_label) { + return None; + } + + if labels.is_empty() { + regs.remove(db_key); + Some(0) + } else { + Some(labels.len()) + } + } + + /// Releases every registration held by `webview_label`, across all databases. + /// + /// Used when a webview window is destroyed without having explicitly called + /// `unobserve()` first - otherwise its registration(s) would leak forever, + /// keeping affected brokers alive indefinitely even though nothing is + /// listening anymore (the "phantom registration" gap noted in the #54 + /// review). Returns the database keys that reached zero remaining observers + /// as a result, so the caller can tear down their brokers. + /// + /// Note: webview labels are reusable across a window's lifetime. A window + /// recreated later with the same static label silently "inherits" whatever + /// registration state is left for that label - harmless for a fixed label + /// like a single main window, but something to be aware of for dynamically + /// labeled windows (e.g. `doc-{id}`), where a fresh window with a *new* + /// label won't be affected by a previous window's leaked registration. + /// + /// `_db_guard` is a witness - see [`register`](Self::register). + pub async fn release_all_for_label( + &self, + _db_guard: &mut DbInstancesGuard<'_>, + webview_label: &str, + ) -> Vec { + let mut regs = self.0.write().await; + let mut newly_empty = Vec::new(); + + regs.retain(|db_key, labels| { + if !labels.remove(webview_label) { + return true; + } + + if labels.is_empty() { + newly_empty.push(db_key.clone()); + false + } else { + true + } + }); + + newly_empty + } + + /// Returns the number of distinct webviews currently registered as observers + /// of `db_key`. Returns `0` if there are none (or the database has never + /// been observed). + /// + /// Test-only: production call sites use the counts already returned by + /// [`register`](Self::register)/[`release`](Self::release) directly, but + /// tests need a way to inspect the current count without mutating it (e.g. + /// to assert an invariant against `DatabaseWrapper::is_observing()` from + /// outside a `register`/`release` call). + #[cfg(test)] + pub async fn count_for_db(&self, db_key: &str) -> usize { + let regs = self.0.read().await; + regs.get(db_key).map_or(0, HashSet::len) + } + + /// Returns whether `webview_label` is currently registered as an observer of + /// `db_key`. + /// + /// Used by `subscribe()` to enforce that the calling webview called + /// `observe()` for `db_key` itself, rather than merely riding along on some + /// other window's registration while a broker happens to exist (see + /// `subscribe()`'s doc comment in `src/commands.rs` and issue #54). + /// + /// Read-only, and deliberately takes no [`DbInstancesGuard`] witness - unlike + /// `register`/`release`/etc., which mutate this registry and need the + /// witness to enforce a lock-acquisition order relative to `db_instances`. A + /// reader has no ordering to prove by itself; it's `subscribe()`'s job to + /// hold `db_instances`'s read lock across both this check and + /// `subscribe_stream()` to make the pair race-free against a concurrent + /// `unobserve()`. + pub(crate) async fn is_registered(&self, db_key: &str, webview_label: &str) -> bool { + let regs = self.0.read().await; + regs + .get(db_key) + .is_some_and(|labels| labels.contains(webview_label)) + } + + /// Clears all observer registrations for a single database. + /// + /// Used when a database is fully closed or removed (`close()`/`remove()`), + /// which tears down observation unconditionally regardless of how many + /// windows had registered. Without this, stale registrations would survive a + /// close/reload cycle and understate how many *new* observers are needed + /// before the broker on the freshly (re)loaded database is torn down again. + /// + /// `_db_guard` is a witness - see [`register`](Self::register). + pub async fn clear_for_db(&self, _db_guard: &mut DbInstancesGuard<'_>, db_key: &str) { + let mut regs = self.0.write().await; + regs.remove(db_key); + } + + /// Clears all observer registrations for every database (app exit / close_all). + /// + /// `_db_guard` is a witness - see [`register`](Self::register). + pub async fn clear_all(&self, _db_guard: &mut DbInstancesGuard<'_>) { + let mut regs = self.0.write().await; + debug!( + "Clearing observer registrations for {} database(s)", + regs.len() + ); + regs.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Test-only: an empty, never-populated `DbInstances`, used purely to + /// obtain real [`DbInstancesGuard`] witnesses (via its own `write()`) for + /// exercising `ObserverRegistrations`'s mutating methods in isolation. + /// Going through `DbInstances::write()` - rather than building a throwaway + /// `RwLock>` directly - is required now that + /// `DbInstancesGuard`'s field is private: there is no + /// other way to construct one, which is the point. + fn dummy_db_lock() -> crate::DbInstances { + crate::DbInstances::default() + } + + #[tokio::test] + async fn test_register_is_additive_across_distinct_labels() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + assert_eq!( + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await, + 1 + ); + assert_eq!( + regs + .register(&mut db_lock.write().await, "MAIN", "window-b") + .await, + 2 + ); + } + + #[tokio::test] + async fn test_register_same_label_twice_is_idempotent() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + assert_eq!( + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await, + 1 + ); + // Same window calling observe() again (e.g. to add more tables) must not + // inflate the refcount. + assert_eq!( + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await, + 1 + ); + } + + #[tokio::test] + async fn test_release_keeps_broker_live_until_last_observer_leaves() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + regs + .register(&mut db_lock.write().await, "MAIN", "window-b") + .await; + + // One of two observers releases: broker must stay live (non-zero remaining). + assert_eq!( + regs + .release(&mut db_lock.write().await, "MAIN", "window-b") + .await, + Some(1) + ); + + // The last observer releases: broker should now be torn down. + assert_eq!( + regs + .release(&mut db_lock.write().await, "MAIN", "window-a") + .await, + Some(0) + ); + } + + #[tokio::test] + async fn test_release_unknown_label_or_db_is_a_noop() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + // Releasing from a db_key with no registrations at all is "never + // registered", not "last observer released" - `None`, not `Some(0)`. + assert_eq!( + regs + .release(&mut db_lock.write().await, "MAIN", "window-a") + .await, + None + ); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + // Releasing a label that was never registered, for a db_key that DOES + // have other registrations, is also a no-op: window-a's registration is + // left untouched. + assert_eq!( + regs + .release(&mut db_lock.write().await, "MAIN", "window-unknown") + .await, + None + ); + assert_eq!(regs.count_for_db("MAIN").await, 1); + } + + #[tokio::test] + async fn test_release_all_for_label_reports_only_databases_that_reached_zero() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + regs + .register(&mut db_lock.write().await, "MAIN", "window-b") + .await; + regs + .register(&mut db_lock.write().await, "OTHER", "window-a") + .await; + + let mut newly_empty = regs + .release_all_for_label(&mut db_lock.write().await, "window-a") + .await; + newly_empty.sort(); + + // MAIN still has window-b, so it must not be reported as newly empty. + // OTHER had only window-a, so it must be. + assert_eq!(newly_empty, vec!["OTHER".to_string()]); + assert_eq!(regs.count_for_db("MAIN").await, 1); + assert_eq!(regs.count_for_db("OTHER").await, 0); + } + + #[tokio::test] + async fn test_release_all_for_label_is_a_noop_for_unregistered_label() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + + assert!( + regs + .release_all_for_label(&mut db_lock.write().await, "window-unknown") + .await + .is_empty() + ); + assert_eq!(regs.count_for_db("MAIN").await, 1); + } + + #[tokio::test] + async fn test_count_for_db_reflects_distinct_observers() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + assert_eq!(regs.count_for_db("MAIN").await, 0); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; // idempotent + regs + .register(&mut db_lock.write().await, "MAIN", "window-b") + .await; + + assert_eq!(regs.count_for_db("MAIN").await, 2); + } + + #[tokio::test] + async fn test_clear_for_db_only_clears_target_database() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + regs + .register(&mut db_lock.write().await, "OTHER", "window-a") + .await; + + regs.clear_for_db(&mut db_lock.write().await, "MAIN").await; + + // MAIN was cleared, so a fresh single registration starts back at 1. + assert_eq!( + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await, + 1 + ); + // OTHER was untouched, so releasing its only observer reaches zero. + assert_eq!( + regs + .release(&mut db_lock.write().await, "OTHER", "window-a") + .await, + Some(0) + ); + } + + #[tokio::test] + async fn test_clear_all_clears_every_database() { + let regs = ObserverRegistrations::default(); + let db_lock = dummy_db_lock(); + + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await; + regs + .register(&mut db_lock.write().await, "OTHER", "window-a") + .await; + + regs.clear_all(&mut db_lock.write().await).await; + + assert_eq!( + regs + .register(&mut db_lock.write().await, "MAIN", "window-a") + .await, + 1 + ); + assert_eq!( + regs + .register(&mut db_lock.write().await, "OTHER", "window-a") + .await, + 1 + ); + } +}