fix: make observation additive and reference-counted per webview (#54)

CHANGELOG.md

@@ -21,13 +21,18 @@ 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
 
 - Added **`Connection`** trait on **`AppHandle`** for Rust-side opens by registration key.
 - **`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.

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

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"

crates/sqlx-sqlite-observer/src/broker.rs

@@ -55,6 +55,7 @@ pub struct ObservationBroker {
    observed_tables: RwLock<HashSet<String>>,
    table_info: RwLock<HashMap<String, TableInfo>>,
    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<I, S>(&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()]);
+   }
+}

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"