feat: add limits on observed table count and subscriptions

observe() and subscribe() accepted unbounded input that could exhaust
memory and spawn unlimited tasks. Add validation: observe() rejects
empty or >100 table lists, subscribe() rejects >100 subscriptions per
database.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pmorris-dev

src/commands.rs

@@ -644,6 +644,16 @@ pub async fn observe(
    tables: Vec<String>,
    config: Option<ObserverConfigParams>,
 ) -> Result<()> {
+   const MAX_OBSERVED_TABLES: usize = 100;
+   const MAX_CHANNEL_CAPACITY: usize = 10_000;
+
+   if tables.is_empty() || tables.len() > MAX_OBSERVED_TABLES {
+      return Err(Error::InvalidConfig(format!(
+         "tables count must be between 1 and {MAX_OBSERVED_TABLES}, got {}",
+         tables.len()
+      )));
+   }
+
    // Abort plugin-level subscription tasks before the crate-level
    // enable_observation() drops the old broker
    active_subs.remove_for_db(&db).await;
@@ -654,8 +664,6 @@ pub async fn observe(
       .get_mut(&db)
       .ok_or_else(|| Error::DatabaseNotLoaded(db.clone()))?;
 
-   const MAX_CHANNEL_CAPACITY: usize = 10_000;
-
    let mut observer_config = sqlx_sqlite_observer::ObserverConfig::new().with_tables(tables);
 
    if let Some(params) = config {
@@ -690,6 +698,13 @@ pub async fn subscribe(
    tables: Vec<String>,
    on_event: Channel<TableChangePayload>,
 ) -> Result<String> {
+   const MAX_SUBSCRIPTIONS_PER_DATABASE: usize = 100;
+
+   let sub_count = active_subs.count_for_db(&db).await;
+   if sub_count >= MAX_SUBSCRIPTIONS_PER_DATABASE {
+      return Err(Error::TooManySubscriptions(MAX_SUBSCRIPTIONS_PER_DATABASE));
+   }
+
    let instances = db_instances.inner.read().await;
 
    let wrapper = instances

src/error.rs

@@ -43,6 +43,10 @@ pub enum Error {
    #[error("cannot load more than {0} databases")]
    TooManyDatabases(usize),
 
+   /// Too many subscriptions for a single database.
+   #[error("cannot create more than {0} subscriptions per database")]
+   TooManySubscriptions(usize),
+
    /// Invalid configuration parameter.
    #[error("invalid configuration: {0}")]
    InvalidConfig(String),
@@ -83,6 +87,7 @@ impl Error {
          Error::DatabaseNotLoaded(_) => "DATABASE_NOT_LOADED".to_string(),
          Error::ObservationNotEnabled(_) => "OBSERVATION_NOT_ENABLED".to_string(),
          Error::TooManyDatabases(_) => "TOO_MANY_DATABASES".to_string(),
+         Error::TooManySubscriptions(_) => "TOO_MANY_SUBSCRIPTIONS".to_string(),
          Error::InvalidConfig(_) => "INVALID_CONFIG".to_string(),
          Error::Other(_) => "ERROR".to_string(),
       }

src/subscriptions.rs

@@ -161,6 +161,12 @@ impl ActiveSubscriptions {
       }
    }
 
+   /// Count active subscriptions for a specific database.
+   pub async fn count_for_db(&self, db_path: &str) -> usize {
+      let subs = self.0.read().await;
+      subs.values().filter(|sub| sub.db_path == db_path).count()
+   }
+
    /// Abort all subscriptions (for cleanup on app exit).
    pub async fn abort_all(&self) {
       let mut subs = self.0.write().await;