fix: cap observer channel_capacity to prevent memory exhaustion

observe() passed user-supplied channel_capacity directly to
tokio::sync::broadcast::channel() without bounds. A capacity of 0
panics, and a very large value pre-allocates an enormous ring buffer.

Add an assertion in ObservationBroker::new() to reject zero capacity at
the crate level, preventing a panic regardless of how the crate is used.
In the plugin layer, validate that channel_capacity is between 1 and
10,000 before applying it, returning an INVALID_CONFIG error otherwise.

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

Author: pmorris-dev

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

@@ -59,7 +59,12 @@ pub struct ObservationBroker {
 
 impl ObservationBroker {
    /// Creates a new broker with the specified broadcast channel capacity.
+   ///
+   /// # Panics
+   ///
+   /// Panics if `channel_capacity` is 0.
    pub fn new(channel_capacity: usize, capture_values: bool) -> Arc<Self> {
+      assert!(channel_capacity > 0, "channel_capacity must be at least 1");
       let (change_tx, _) = broadcast::channel(channel_capacity);
       Arc::new(Self {
          buffer: Mutex::new(Vec::new()),

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

@@ -90,6 +90,9 @@ impl ObserverConfig {
 
    /// Sets the broadcast channel capacity for change notifications.
    ///
+   /// Capacity must be at least 1. A capacity of 0 will cause a panic when the
+   /// observer is initialized.
+   ///
    /// See [`channel_capacity`](Self::channel_capacity) for details on sizing.
    pub fn with_channel_capacity(mut self, capacity: usize) -> Self {
       self.channel_capacity = capacity;

src/commands.rs

@@ -647,10 +647,17 @@ 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 {
       if let Some(capacity) = params.channel_capacity {
+         if capacity == 0 || capacity > MAX_CHANNEL_CAPACITY {
+            return Err(Error::InvalidConfig(format!(
+               "channel_capacity must be between 1 and {MAX_CHANNEL_CAPACITY}, got {capacity}"
+            )));
+         }
          observer_config = observer_config.with_channel_capacity(capacity);
       }
       if let Some(capture) = params.capture_values {

src/error.rs

@@ -39,6 +39,10 @@ pub enum Error {
    #[error("observation not enabled for database: {0}")]
    ObservationNotEnabled(String),
 
+   /// Invalid configuration parameter.
+   #[error("invalid configuration: {0}")]
+   InvalidConfig(String),
+
    /// Generic error for operations that don't fit other categories.
    #[error("{0}")]
    Other(String),
@@ -74,6 +78,7 @@ impl Error {
          Error::PathTraversal(_) => "PATH_TRAVERSAL".to_string(),
          Error::DatabaseNotLoaded(_) => "DATABASE_NOT_LOADED".to_string(),
          Error::ObservationNotEnabled(_) => "OBSERVATION_NOT_ENABLED".to_string(),
+         Error::InvalidConfig(_) => "INVALID_CONFIG".to_string(),
          Error::Other(_) => "ERROR".to_string(),
       }
    }