From ebab6abcf033a490667c9fe72ceaed1a27e42800 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Wed, 10 Jun 2026 10:26:11 -0700 Subject: [PATCH] feat(kv): add watch_prefixes (single multi-filter consumer); v0.3.2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit KvWatcher::watch_prefixes(&[prefix]) watches the union of several prefixes through ONE NATS consumer, using server 2.10 multi-filter consumers (async-nats watch_many -> OrderedConfig.filter_subjects), instead of one consumer per prefix. This matters because JetStream consumers are a per-stream resource. Measured on a single stream: ~tens of KB of server RSS per consumer, growing super-linearly past a few thousand (1.5 -> 22 -> 33 -> 56 KB/consumer at 200/600/1200/2400). A node that watches K prefixes must therefore cost 1 consumer, not K — at fleet scale K-per-node is multiple TB of consumer state on one stream leader; 1-per-node is the only shape that fits. Proven by an integration test against a real nats-server (watch_prefixes_unions_on_a_single_consumer): the KV stream carries exactly one consumer, both watched prefixes are delivered, and a third prefix is filtered server-side (never delivered). Empty prefixes watch nothing (NOT the whole bucket). Trait default is required; all impls updated. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 2 +- Cargo.toml | 2 +- benches/applied.rs | 8 +++++++ src/applied.rs | 20 ++++++++++++++++++ src/kv.rs | 11 ++++++++++ src/nats.rs | 24 +++++++++++++++++++++ tests/integration.rs | 50 ++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 115 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 702c283..c52448f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -97,7 +97,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.3.1" +version = "0.3.2" dependencies = [ "async-nats", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 0a63110..e3031b1 100644 --- a/Cargo.toml +++ b/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 diff --git a/benches/applied.rs b/benches/applied.rs index 63e26c7..029dfa0 100644 --- a/benches/applied.rs +++ b/benches/applied.rs @@ -39,6 +39,14 @@ impl KvWatcher for ScriptedWatcher { async fn watch_prefix(&self, _prefix: &str, tx: Sender) -> Result<(), KvError> { self.watch_all(tx).await } + + async fn watch_prefixes( + &self, + _prefixes: &[&str], + tx: Sender, + ) -> Result<(), KvError> { + self.watch_all(tx).await + } } fn put(i: u64) -> KvUpdate { diff --git a/src/applied.rs b/src/applied.rs index 026d3c8..3f31bd4 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -465,6 +465,18 @@ mod tests { Ok(()) } + async fn watch_prefixes( + &self, + _prefixes: &[&str], + tx: Sender, + ) -> 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) -> Result<(), KvError> { Err(KvError::WatchError("injected watch failure".into())) } + + async fn watch_prefixes( + &self, + _prefixes: &[&str], + _tx: Sender, + ) -> Result<(), KvError> { + Err(KvError::WatchError("injected watch failure".into())) + } } // A no-op parse that keeps every Put as the value bytes; drops deletes. diff --git a/src/kv.rs b/src/kv.rs index 649154e..6f6dc4f 100644 --- a/src/kv.rs +++ b/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) -> 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) + -> Result<(), KvError>; + /// Resume watching all keys from a previously saved cursor position. /// /// Returns `KvError::CursorExpired` if the backend has compacted past the diff --git a/src/nats.rs b/src/nats.rs index f3980f9..36cf32a 100644 --- a/src/nats.rs +++ b/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, + ) -> 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 = 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, diff --git a/tests/integration.rs b/tests/integration.rs index 74e05c5..a46b72d 100644 --- a/tests/integration.rs +++ b/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 = + 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;