From abb800121d1c73963836f77407126c0f51e65752 Mon Sep 17 00:00:00 2001 From: Jordan Hafer <42755763+jjhafer@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:50:19 -0400 Subject: [PATCH] fix: make observation additive and reference-counted per webview (#54) Observation was destructive: any observe() call dropped the existing broker, and with it the sole broadcast sender, ending every current subscriber's stream with no error and no attributable cause. Since observe() is exposed over IPC, one window could silently kill another window's change stream just by observing the same database, which made multi-window use impossible. Observation is now shared rather than replaced. The broker is created once per database and reused, requested tables are merged into it, and each window holds its own registration so teardown waits for the last one to be released. Correctness depends on the registrations and the broker staying in step, so the lock order protecting them is enforced by the compiler rather than by convention - the registration methods require proof that the caller holds the database lock. The same ordering defect had been reintroduced four times while the rule lived only in a comment. Fixes issue #54. BREAKING CHANGE: DatabaseWrapper::enable_observation() no longer tears down the existing broker. 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. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 19 + Cargo.lock | 4 +- README.md | 41 +- crates/sqlx-sqlite-observer/Cargo.toml | 2 +- crates/sqlx-sqlite-observer/src/broker.rs | 73 + crates/sqlx-sqlite-toolkit/Cargo.toml | 2 +- crates/sqlx-sqlite-toolkit/src/wrapper.rs | 56 +- .../tests/observation_tests.rs | 147 ++ guest-js/index.ts | 54 +- src/commands.rs | 413 ++++- src/error.rs | 6 + src/lib.rs | 1606 ++++++++++++++++- src/subscriptions.rs | 542 +++++- 13 files changed, 2906 insertions(+), 59 deletions(-) create mode 100644 crates/sqlx-sqlite-toolkit/tests/observation_tests.rs 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 + ); + } +}