feat(kv): watch_prefixes — single multi-filter consumer (v0.3.2)

Cargo.toml

@@ -8,7 +8,7 @@ rust-version = "1.92"
 
 [package]
 name = "beyond-slipstream"
-version = "0.3.1"
+version = "0.3.2"
 edition.workspace = true
 license.workspace = true
 rust-version.workspace = true

benches/applied.rs

@@ -39,6 +39,14 @@ impl KvWatcher for ScriptedWatcher {
     async fn watch_prefix(&self, _prefix: &str, tx: Sender<KvUpdate>) -> Result<(), KvError> {
         self.watch_all(tx).await
     }
+
+    async fn watch_prefixes(
+        &self,
+        _prefixes: &[&str],
+        tx: Sender<KvUpdate>,
+    ) -> Result<(), KvError> {
+        self.watch_all(tx).await
+    }
 }
 
 fn put(i: u64) -> KvUpdate {

src/applied.rs

@@ -465,6 +465,18 @@ mod tests {
             Ok(())
         }
 
+        async fn watch_prefixes(
+            &self,
+            _prefixes: &[&str],
+            tx: Sender<KvUpdate>,
+        ) -> Result<(), KvError> {
+            // This mock scripts the applied-watch resumption tests, not prefix
+            // filtering; it delivers the same `full` script as `watch_prefix`.
+            // The real multi-filter scoping is proved in the NATS integration test.
+            self.deliver(&self.full, tx).await;
+            Ok(())
+        }
+
         async fn watch_all_from(
             &self,
             _cursor: &WatchCursor,
@@ -509,6 +521,14 @@ mod tests {
         async fn watch_prefix(&self, _prefix: &str, _tx: Sender<KvUpdate>) -> Result<(), KvError> {
             Err(KvError::WatchError("injected watch failure".into()))
         }
+
+        async fn watch_prefixes(
+            &self,
+            _prefixes: &[&str],
+            _tx: Sender<KvUpdate>,
+        ) -> Result<(), KvError> {
+            Err(KvError::WatchError("injected watch failure".into()))
+        }
     }
 
     // A no-op parse that keeps every Put as the value bytes; drops deletes.

src/kv.rs

@@ -261,6 +261,17 @@ pub trait KvWatcher: Send + Sync {
     /// Watch keys matching a prefix.
     async fn watch_prefix(&self, prefix: &str, tx: Sender<KvUpdate>) -> Result<(), KvError>;
 
+    /// Watch keys matching ANY of `prefixes`, delivered through one channel.
+    ///
+    /// The contract is exactly the union of the prefixes — no other keys. A
+    /// backend with native multi-filter consumers (NATS server 2.10+) serves all
+    /// `prefixes` from a SINGLE consumer; that matters because consumers are a
+    /// per-stream resource (measured at ~tens of KB of server state each, growing
+    /// super-linearly past a few thousand on one stream), so a watcher scoped to N
+    /// prefixes must not cost N consumers.
+    async fn watch_prefixes(&self, prefixes: &[&str], tx: Sender<KvUpdate>)
+    -> Result<(), KvError>;
+
     /// Resume watching all keys from a previously saved cursor position.
     ///
     /// Returns `KvError::CursorExpired` if the backend has compacted past the

src/nats.rs

@@ -1047,6 +1047,30 @@ impl KvWatcher for NatsKvWatcher {
         stream_watch(watcher, &tx).await
     }
 
+    async fn watch_prefixes(
+        &self,
+        prefixes: &[&str],
+        tx: Sender<KvUpdate>,
+    ) -> Result<(), KvError> {
+        if prefixes.is_empty() {
+            // Nothing to watch. Critically, do NOT fall through to `watch_many`
+            // with an empty filter set — an unfiltered ordered consumer would
+            // watch the WHOLE bucket, the opposite of a scoped watch.
+            return Ok(());
+        }
+        // ONE multi-filter consumer for every prefix (NATS 2.10 `filter_subjects`)
+        // rather than one consumer per prefix. `watch_many` builds a single ordered
+        // push consumer with `filter_subjects = [{p}> ...]` and yields the same
+        // `Entry` stream as `watch`, so `stream_watch` is reused verbatim. This is
+        // the per-stream-consumer-count fix: a node scoped to N prefixes costs 1
+        // consumer, not N.
+        let keys: Vec<String> = prefixes.iter().map(|p| format!("{p}>")).collect();
+        let watcher = timed(self.kv.watch_many(keys))
+            .await?
+            .map_err(|e| KvError::WatchError(e.to_string()))?;
+        stream_watch(watcher, &tx).await
+    }
+
     async fn watch_all_from(
         &self,
         cursor: &WatchCursor,

tests/integration.rs

@@ -569,6 +569,56 @@ async fn watch_prefix_filters_by_subject() {
     assert!(matches!(&updates[0], KvUpdate::Put(e) if e.key == "node.a"));
 }
 
+/// `watch_prefixes` must serve the UNION of several prefixes from a SINGLE
+/// multi-filter consumer (NATS 2.10 `filter_subjects`), not one consumer per
+/// prefix — because consumers are a per-stream resource (~tens of KB of server
+/// state each, super-linear past a few thousand on one stream). Proven two ways:
+/// the KV stream carries exactly one consumer, and a non-watched prefix is
+/// filtered server-side (never delivered), while both watched prefixes are.
+#[tokio::test]
+async fn watch_prefixes_unions_on_a_single_consumer() {
+    let nats = TestNats::start().await;
+    let (_conn, store) = nats.store("watchmany").await;
+    let writer = store.writer().expect("writer");
+    let watcher = store.watcher().expect("watcher");
+
+    let (tx, mut rx) = mpsc::channel(64);
+    tokio::spawn(async move {
+        let _ = watcher.watch_prefixes(&["vpcA.", "vpcB."], tx).await;
+    });
+    // Sentinel must fall within one watched prefix to be echoed back.
+    establish_watch(writer.as_ref(), &mut rx, "vpcA.__ready__").await;
+
+    // PROOF 1 — one multi-filter consumer, not N. The KV bucket `watchmany`
+    // is JetStream stream `KV_watchmany`; assert it carries exactly one consumer.
+    let js = async_nats::jetstream::new(async_nats::connect(&nats.url).await.expect("raw connect"));
+    let mut stream = js.get_stream("KV_watchmany").await.expect("get kv stream");
+    let info = stream.info().await.expect("stream info");
+    assert_eq!(
+        info.state.consumer_count, 1,
+        "watch_prefixes must use ONE multi-filter consumer, found {}",
+        info.state.consumer_count
+    );
+
+    // PROOF 2 — union delivered, third prefix filtered server-side.
+    writer.put("vpcC.x", b"skip").await.expect("put vpcC"); // outside the union
+    writer.put("vpcA.a", b"keep").await.expect("put vpcA");
+    writer.put("vpcB.b", b"keep").await.expect("put vpcB");
+
+    let updates = collect_updates(&mut rx, 2).await;
+    let keys: std::collections::HashSet<String> =
+        updates.iter().map(|u| u.key().to_string()).collect();
+    assert!(
+        keys.contains("vpcA.a") && keys.contains("vpcB.b"),
+        "both watched prefixes must be delivered, got {keys:?}"
+    );
+    // vpcC.x must NEVER arrive — server-side filtered, not client-discarded.
+    assert!(
+        timeout(Duration::from_millis(500), rx.recv()).await.is_err(),
+        "a non-watched prefix leaked through watch_prefixes"
+    );
+}
+
 #[tokio::test]
 async fn watch_all_from_replays_only_the_delta() {
     let nats = TestNats::start().await;