sub(feat): allow user to handle stream lag

- Previously we were just logging a warning when
the number of changes in a tx exceeded the channel
capacity. Now we report the lag in the change event
to make it more obvious to the user and allow for the
user to respond to this "error" condition.

Author: pmorris-dev

crates/sqlx-sqlite-observer/README.md

@@ -95,6 +95,8 @@ This ensures subscribers **only receive notifications for committed changes**.
 ### Core Types
 
    * **`TableChange`**: Notification of a change to a database table
+   * **`TableChangeEvent`**: Event yielded by `TableChangeStream` —
+     either `Change(TableChange)` or `Lagged(u64)`
    * **`ChangeOperation`**: Insert, Update, or Delete
    * **`ColumnValue`**: Typed column value (Null, Integer, Real, Text, Blob)
    * **`ObserverConfig`**: Configuration for table filtering and channel
@@ -111,7 +113,7 @@ This ensures subscribers **only receive notifications for committed changes**.
    * **`TableChangeStreamExt`**: Extension trait for converting receivers to
      streams
 
-### SQLx SQLite Connection Manager Integration (feature: `conn-mgr`) <!-- COMING SOON -->
+### SQLx SQLite Connection Manager Integration (feature: `conn-mgr`)
 
    * **`ObservableSqliteDatabase`**: Wrapper for `SqliteDatabase` with observation
    * **`ObservableWriteGuard`**: Write guard with hooks registered
@@ -179,7 +181,7 @@ meaningful/correct for non-integer or composite primary keys.
 
 ### Basic Usage
 
-```rust,no_run
+```rust
 use sqlx::SqlitePool;
 use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig, ColumnValue};
 
@@ -219,10 +221,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
 ### Stream API
 
-```rust,no_run
+```rust
 use futures::StreamExt;
 use sqlx::SqlitePool;
-use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig};
+use sqlx_sqlite_observer::{
+    SqliteObserver, ObserverConfig, TableChangeEvent,
+};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -232,13 +236,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let mut stream = observer.subscribe_stream(["users"]);
 
-    while let Some(change) = stream.next().await {
-        println!(
-            "Table {} row {} was {:?}",
-            change.table,
-            change.rowid.unwrap_or(-1),
-            change.operation
-        );
+    while let Some(event) = stream.next().await {
+        match event {
+            TableChangeEvent::Change(change) => {
+                println!(
+                    "Table {} row {} was {:?}",
+                    change.table,
+                    change.rowid.unwrap_or(-1),
+                    change.operation
+                );
+            }
+            TableChangeEvent::Lagged(n) => {
+                eprintln!("Missed {} notifications", n);
+            }
+        }
     }
 
     Ok(())
@@ -247,7 +258,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
 ### Value Capture
 
-```rust,no_run
+```rust
 use sqlx::SqlitePool;
 use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig, ColumnValue};
 
@@ -284,7 +295,37 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
 ### SQLx SQLite Connection Manager Integration
 
-<!-- TODO: Add example showing ObservableSqliteDatabase usage -->
+```rust
+use std::sync::Arc;
+use sqlx_sqlite_conn_mgr::SqliteDatabase;
+use sqlx_sqlite_observer::{
+    ObservableSqliteDatabase, ObserverConfig,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let db = SqliteDatabase::connect("mydb.db", None).await?;
+    let config = ObserverConfig::new().with_tables(["users"]);
+    let observable = ObservableSqliteDatabase::new(db, config);
+
+    let mut rx = observable.subscribe(["users"]);
+
+    // Write through the observable writer
+    let mut writer = observable.acquire_writer().await?;
+    sqlx::query("BEGIN").execute(&mut *writer).await?;
+    sqlx::query("INSERT INTO users (name) VALUES (?)")
+        .bind("Alice")
+        .execute(&mut *writer)
+        .await?;
+    sqlx::query("COMMIT").execute(&mut *writer).await?;
+
+    // Notification arrives after commit
+    let change = rx.recv().await?;
+    println!("Changed: {}", change.table);
+
+    Ok(())
+}
+```
 
 ## Usage Notes
 
@@ -301,6 +342,58 @@ let config = ObserverConfig::new()
     .with_channel_capacity(1000); // Handle large transactions
 ```
 
+### Handling Lag
+
+When using the Stream API, the stream yields `TableChangeEvent` values.
+Most events are `Change` variants, but if a consumer falls behind, the
+stream yields a `Lagged(n)` event indicating how many notifications
+were missed.
+
+```rust
+use futures::StreamExt;
+use sqlx_sqlite_observer::TableChangeEvent;
+# use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig};
+# async fn example(observer: SqliteObserver) {
+
+let mut stream = observer.subscribe_stream(["users"]);
+
+while let Some(event) = stream.next().await {
+    match event {
+        TableChangeEvent::Change(change) => {
+            // Process the change normally
+        }
+        TableChangeEvent::Lagged(n) => {
+            // n notifications were missed — local state may be stale.
+            // Re-query the database for current state.
+            tracing::warn!("Missed {} change notifications", n);
+        }
+    }
+}
+# }
+```
+
+**When does lag happen?** The broadcast channel has a fixed capacity
+(default 256). Lag occurs when the oldest unread messages are
+overwritten. This can happen in two ways:
+
+   * A subscriber processes changes slower than they arrive
+   * A single transaction contains more mutating statements than the
+     channel capacity, causing messages to be overwritten before the
+     consumer reads them
+
+This is rare under normal conditions but can occur during bulk
+writes or large transactions.
+
+**How to prevent it:**
+
+   * Increase `channel_capacity` via `ObserverConfig::with_channel_capacity`
+   * Process changes faster (avoid blocking in the stream consumer)
+   * Use a dedicated task for stream consumption
+
+**Note:** The `broadcast::Receiver` API (from `subscribe()`) surfaces
+lag as `RecvError::Lagged(n)` — the same information, just through
+the raw tokio broadcast channel interface rather than the stream.
+
 ### Disabling Value Capture
 
 By default, `TableChange` includes `old_values` and `new_values` with the actual

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

@@ -100,6 +100,30 @@ impl ColumnValue {
    }
 }
 
+/// Event yielded by [`TableChangeStream`](crate::stream::TableChangeStream).
+///
+/// Most events are `Change` variants containing the actual table change data.
+/// A `Lagged` event indicates the consumer fell behind and missed some
+/// notifications — consider increasing
+/// [`channel_capacity`](crate::config::ObserverConfig::channel_capacity).
+#[derive(Debug, Clone)]
+pub enum TableChangeEvent {
+   /// A table change notification.
+   Change(TableChange),
+   /// The stream fell behind and missed `n` change notifications.
+   ///
+   /// This can happen when:
+   /// - The consumer is processing changes too slowly relative to the
+   ///   rate of database writes.
+   /// - A single transaction contains more mutating statements than the
+   ///   channel capacity, causing older messages to be overwritten before
+   ///   the consumer reads them.
+   ///
+   /// When this happens, the consumer should assume its local state may
+   /// be stale and re-query the database for the current state.
+   Lagged(u64),
+}
+
 /// Notification of a change to a database table.
 ///
 /// Contains the table name, operation type, affected rowid, and the

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

@@ -81,7 +81,9 @@
 //! ```rust,no_run
 //! use futures::StreamExt;
 //! use sqlx::SqlitePool;
-//! use sqlx_sqlite_observer::{ChangeOperation, SqliteObserver, ObserverConfig, TableChangeStreamExt};
+//! use sqlx_sqlite_observer::{
+//!     SqliteObserver, ObserverConfig, TableChangeEvent, TableChangeStreamExt,
+//! };
 //!
 //! #[tokio::main]
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -92,20 +94,26 @@
 //!     // Get a Stream instead of broadcast::Receiver
 //!     let mut stream = observer.subscribe_stream(["users"]);
 //!
-//!     // Use standard Stream combinators
-//!     while let Some(change) = stream.next().await {
-//!         println!(
-//!             "Table {} row {} was {:?}",
-//!             change.table,
-//!             change.rowid.unwrap_or(-1),
-//!             change.operation
-//!         );
-//!         // Access typed column values
-//!         if let Some(old) = &change.old_values {
-//!             println!("  Old values: {:?}", old);
-//!         }
-//!         if let Some(new) = &change.new_values {
-//!             println!("  New values: {:?}", new);
+//!     // Stream yields TableChangeEvent variants
+//!     while let Some(event) = stream.next().await {
+//!         match event {
+//!             TableChangeEvent::Change(change) => {
+//!                 println!(
+//!                     "Table {} row {} was {:?}",
+//!                     change.table,
+//!                     change.rowid.unwrap_or(-1),
+//!                     change.operation
+//!                 );
+//!                 if let Some(old) = &change.old_values {
+//!                     println!("  Old values: {:?}", old);
+//!                 }
+//!                 if let Some(new) = &change.new_values {
+//!                     println!("  New values: {:?}", new);
+//!                 }
+//!             }
+//!             TableChangeEvent::Lagged(n) => {
+//!                 eprintln!("Missed {} notifications, re-query state", n);
+//!             }
 //!         }
 //!     }
 //!
@@ -127,7 +135,7 @@ pub mod stream;
 pub mod conn_mgr;
 
 pub use broker::ObservationBroker;
-pub use change::{ChangeOperation, ColumnValue, TableChange, TableInfo};
+pub use change::{ChangeOperation, ColumnValue, TableChange, TableChangeEvent, TableInfo};
 pub use config::ObserverConfig;
 pub use connection::ObservableConnection;
 pub use error::Error;

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

@@ -6,7 +6,7 @@ use tokio_stream::Stream;
 use tokio_stream::wrappers::BroadcastStream;
 use tracing::warn;
 
-use crate::change::TableChange;
+use crate::change::{TableChange, TableChangeEvent};
 
 /// A filtered stream of table change notifications.
 ///
@@ -32,7 +32,7 @@ impl TableChangeStream {
 }
 
 impl Stream for TableChangeStream {
-   type Item = TableChange;
+   type Item = TableChangeEvent;
 
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
       loop {
@@ -46,15 +46,17 @@ impl Stream for TableChangeStream {
                {
                   continue;
                }
-               return Poll::Ready(Some(change));
+               return Poll::Ready(Some(TableChangeEvent::Change(change)));
             }
-            Poll::Ready(Some(Err(err))) => {
-               // Lagged error - missed some messages due to slow consumption
+            Poll::Ready(Some(Err(
+               tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(count),
+            ))) => {
                warn!(
-                  error = %err,
-                  "Stream lagged — missed change notifications. Consider increasing channel_capacity."
+                  missed = count,
+                  "Stream lagged — missed change notifications. \
+                   Consider increasing channel_capacity."
                );
-               continue;
+               return Poll::Ready(Some(TableChangeEvent::Lagged(count)));
             }
             Poll::Ready(None) => return Poll::Ready(None),
             Poll::Pending => return Poll::Pending,

crates/sqlx-sqlite-observer/tests/conn_mgr_tests.rs

@@ -331,6 +331,13 @@ async fn test_stream_receives_notifications() {
    let result = timeout(Duration::from_millis(100), stream.next()).await;
    assert!(result.is_ok(), "Stream receives notification");
 
-   let change = result.unwrap().unwrap();
-   assert_eq!(change.table, "users");
+   let event = result.unwrap().unwrap();
+   match event {
+      sqlx_sqlite_observer::TableChangeEvent::Change(change) => {
+         assert_eq!(change.table, "users");
+      }
+      sqlx_sqlite_observer::TableChangeEvent::Lagged(_) => {
+         panic!("Expected Change event, got Lagged");
+      }
+   }
 }

crates/sqlx-sqlite-observer/tests/integration_tests.rs

@@ -391,8 +391,15 @@ async fn test_stream_receives_notifications() {
    let result = timeout(Duration::from_millis(100), stream.next()).await;
    assert!(result.is_ok(), "Stream receives notification");
 
-   let change = result.unwrap().unwrap();
-   assert_eq!(change.table, "users");
+   let event = result.unwrap().unwrap();
+   match event {
+      sqlx_sqlite_observer::TableChangeEvent::Change(change) => {
+         assert_eq!(change.table, "users");
+      }
+      sqlx_sqlite_observer::TableChangeEvent::Lagged(_) => {
+         panic!("Expected Change event, got Lagged");
+      }
+   }
 }
 
 #[tokio::test]
@@ -424,6 +431,50 @@ async fn test_stream_filters_tables() {
    assert!(result.is_err(), "Stream filters out non-subscribed tables");
 }
 
+#[tokio::test]
+async fn test_stream_lag_when_capacity_exceeded() {
+   let pool = setup_test_db().await;
+   let config = ObserverConfig::new()
+      .with_tables(["users"])
+      .with_channel_capacity(2);
+
+   let observer = SqliteObserver::new(pool, config);
+
+   let mut stream = observer.subscribe_stream(["users"]);
+   let mut conn = observer.acquire().await.unwrap();
+
+   // Insert more rows than the channel capacity in a single transaction
+   sqlx::query("BEGIN").execute(&mut **conn).await.unwrap();
+   for i in 0..5 {
+      sqlx::query("INSERT INTO users (name) VALUES (?)")
+         .bind(format!("User{}", i))
+         .execute(&mut **conn)
+         .await
+         .unwrap();
+   }
+   sqlx::query("COMMIT").execute(&mut **conn).await.unwrap();
+
+   let mut saw_lagged = false;
+   let mut saw_change = false;
+
+   // Drain all available events
+   while let Ok(Some(event)) = timeout(Duration::from_millis(100), stream.next()).await {
+      match event {
+         sqlx_sqlite_observer::TableChangeEvent::Lagged(n) => {
+            assert!(n > 0, "Lagged count should be > 0");
+            saw_lagged = true;
+         }
+         sqlx_sqlite_observer::TableChangeEvent::Change(change) => {
+            assert_eq!(change.table, "users");
+            saw_change = true;
+         }
+      }
+   }
+
+   assert!(saw_lagged, "Expected at least one Lagged event");
+   assert!(saw_change, "Expected at least one Change event");
+}
+
 // ============================================================================
 // Value Capture
 // ============================================================================