proxima_primitives/pipe/
live_filter.rs

1//! [`LiveFilter`] — a decision pipe whose predicate is swapped live through a
2//! control handle. The pub/sub filter whose membership changes at runtime
3//! without tearing down the subscription.
4//!
5//! It is a thin [`SendPipe`] face over [`proxima_core::live::Live`]: the general
6//! "hot readers, rare control-plane writes" cell, specialised to answer
7//! pass/reject (`Ok(item)` = admit, `Err(item)` = reject — see
8//! `fan_in.rs`'s module doc for why a data-path decision is a pipe, not a
9//! `bool`-returning capability trait). [`live_filter`] splits into the two
10//! shared-state halves:
11//!
12//! - [`LiveFilter`] — the filtering (read) half. It implements [`SendPipe`], so
13//!   it drops straight into `.and_then(inner)`; the hot data path does one
14//!   lock-free load and delegates to the current predicate. For a pure
15//!   membership query that never forwards the item on (a control-plane
16//!   question, not a data-path decision — [`FilterRegistry::matches`](crate::pipe::filter_registry::FilterRegistry::matches)'s
17//!   fan-out lookup is the example), [`LiveFilter::contains`] answers
18//!   synchronously without the pipe detour.
19//! - [`FilterControl`] — the control (write) half. The "pipe pumping in":
20//!   [`replace`](FilterControl::replace) / [`update`](FilterControl::update)
21//!   swap the predicate, and for id subscriptions
22//!   [`apply`](FilterControl::apply) folds a [`FilterUpdate`] into it. Cheap to
23//!   clone; drive it directly or from a stream of updates drained through an
24//!   [`UnpinPipe`](crate::pipe::UnpinPipe) source.
25//!
26//! Both halves (and every clone) share one cell, so a control mutation is
27//! visible to all filter clones — this is the "split" of a pub/sub subscription:
28//! data-out ([`LiveFilter`]) and control-in ([`FilterControl`]) are the two ends
29//! of one object. The swap is per-call monotonic (see
30//! [`proxima_core::live`] for the ordering contract): a call
31//! sees either the pre-swap or post-swap predicate, which is the right semantics
32//! for id filtering ("start matching id X soon after I add it"). A filter that
33//! must change at a precise item needs the update carried *in* the item stream
34//! (a [`FanIn`](crate::pipe::FanIn) select over data + control) instead.
35
36use alloc::collections::BTreeSet;
37use alloc::sync::Arc;
38use core::future::Future;
39
40use proxima_core::live::{Live, LiveControl, live};
41use serde::{Deserialize, Serialize};
42
43use crate::pipe::SendPipe;
44
45/// The filtering (read) half of a [`live_filter`] split: a [`SendPipe`] whose
46/// predicate is swapped live by the paired [`FilterControl`]. Clone is a single
47/// `Arc` bump — hand a clone to every subscriber.
48pub struct LiveFilter<Predicate> {
49    inner: Live<Predicate>,
50}
51
52/// The control (write) half of a [`live_filter`] split — the "pipe pumping in"
53/// that swaps the predicate the paired [`LiveFilter`] reads.
54pub struct FilterControl<Predicate> {
55    control: LiveControl<Predicate>,
56}
57
58/// Split an initial predicate into a live filtering half and its control half,
59/// sharing one cell. `initial` is the only input (e.g.
60/// `IdSet::from_ids(config.subscribed_ids)`).
61#[must_use]
62pub fn live_filter<Predicate>(
63    initial: Predicate,
64) -> (LiveFilter<Predicate>, FilterControl<Predicate>) {
65    let (inner, control) = live(initial);
66    (LiveFilter { inner }, FilterControl { control })
67}
68
69/// Sugar for the id-subscription case: split a live [`IdSet`] filter seeded from
70/// `ids`. Equivalent to `live_filter(IdSet::from_ids(ids))`.
71#[must_use]
72pub fn live_filter_ids<Id: Ord + Clone>(
73    ids: impl IntoIterator<Item = Id>,
74) -> (LiveFilter<IdSet<Id>>, FilterControl<IdSet<Id>>) {
75    live_filter(IdSet::from_ids(ids))
76}
77
78impl<Predicate> Clone for LiveFilter<Predicate> {
79    fn clone(&self) -> Self {
80        Self {
81            inner: self.inner.clone(),
82        }
83    }
84}
85
86impl<Predicate> LiveFilter<Predicate> {
87    /// Snapshot the current predicate (introspection).
88    #[must_use]
89    pub fn snapshot(&self) -> Arc<Predicate> {
90        self.inner.snapshot()
91    }
92}
93
94impl<Id: Ord> LiveFilter<IdSet<Id>> {
95    /// Synchronous membership check — for a control-plane query that only
96    /// asks "does `id` match" and never forwards the item onward (unlike the
97    /// [`SendPipe`] form, which admits or rejects an item flowing through it).
98    #[must_use]
99    pub fn contains(&self, id: &Id) -> bool {
100        self.inner.read(|set| set.contains(id))
101    }
102}
103
104impl<Item, Predicate> SendPipe for LiveFilter<Predicate>
105where
106    Predicate: SendPipe<In = Item, Out = Item> + Send + Sync + 'static,
107    Item: Send + 'static,
108{
109    type In = Item;
110    type Out = Item;
111    type Err = Predicate::Err;
112
113    fn call(&self, input: Item) -> impl Future<Output = Result<Item, Predicate::Err>> + Send {
114        // snapshot outside the lock, then await the call — `Live::read`'s
115        // closure is synchronous and cannot itself hold an in-flight future.
116        let predicate = self.snapshot();
117        async move { predicate.call(input).await }
118    }
119}
120
121impl<Predicate> Clone for FilterControl<Predicate> {
122    fn clone(&self) -> Self {
123        Self {
124            control: self.control.clone(),
125        }
126    }
127}
128
129impl<Predicate> FilterControl<Predicate> {
130    /// Replace the predicate wholesale. Every subsequent read of the paired
131    /// [`LiveFilter`] sees `next`.
132    pub fn replace(&self, next: Predicate) {
133        self.control.replace(next);
134    }
135
136    /// Read-modify-write the predicate. `mutate` derives the next predicate from
137    /// the current one and may be retried under contention, so it must be pure.
138    pub fn update(&self, mutate: impl Fn(&Predicate) -> Predicate) {
139        self.control.update(mutate);
140    }
141
142    /// Snapshot the current predicate (introspection).
143    #[must_use]
144    pub fn snapshot(&self) -> Arc<Predicate> {
145        self.control.snapshot()
146    }
147}
148
149/// A membership predicate over a set of ids — the canonical pub/sub filter.
150/// [`contains`](Self::contains) is `true` when the set holds `id`; as a
151/// [`SendPipe`] it admits (`Ok(id)`) or rejects (`Err(id)`) the id flowing
152/// through it. Build one as the `initial` for [`live_filter`] and grow it
153/// live with [`FilterUpdate`].
154///
155/// Serde-transparent: it round-trips as a bare array of ids, so a subscription's
156/// set is exactly what you load from config or ship over the wire (the config
157/// mirror; the layered `conflaguration` loader attaches at the service that
158/// composes this, per the workspace config principle's layering caveat).
159#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
160#[serde(transparent)]
161pub struct IdSet<Id: Ord> {
162    members: BTreeSet<Id>,
163}
164
165impl<Id: Ord> IdSet<Id> {
166    /// An empty set — matches nothing.
167    #[must_use]
168    pub fn new() -> Self {
169        Self {
170            members: BTreeSet::new(),
171        }
172    }
173
174    /// A set seeded from `ids`.
175    pub fn from_ids(ids: impl IntoIterator<Item = Id>) -> Self {
176        ids.into_iter().collect()
177    }
178
179    /// Whether `id` is a member.
180    #[must_use]
181    pub fn contains(&self, id: &Id) -> bool {
182        self.members.contains(id)
183    }
184
185    /// The number of ids in the set.
186    #[must_use]
187    pub fn len(&self) -> usize {
188        self.members.len()
189    }
190
191    /// Whether the set is empty (matches nothing).
192    #[must_use]
193    pub fn is_empty(&self) -> bool {
194        self.members.is_empty()
195    }
196
197    /// Iterate the member ids in sorted order.
198    pub fn iter(&self) -> impl Iterator<Item = &Id> + '_ {
199        self.members.iter()
200    }
201}
202
203impl<Id: Ord + Clone> IdSet<Id> {
204    /// A copy with `id` added — the copy-on-write step for
205    /// [`FilterControl::update`].
206    #[must_use]
207    pub fn with(&self, id: Id) -> Self {
208        let mut members = self.members.clone();
209        members.insert(id);
210        Self { members }
211    }
212
213    /// A copy with `id` removed.
214    #[must_use]
215    pub fn without(&self, id: &Id) -> Self {
216        let mut members = self.members.clone();
217        members.remove(id);
218        Self { members }
219    }
220}
221
222impl<Id: Ord> Default for IdSet<Id> {
223    fn default() -> Self {
224        Self::new()
225    }
226}
227
228impl<Id: Ord> FromIterator<Id> for IdSet<Id> {
229    fn from_iter<Iter: IntoIterator<Item = Id>>(iter: Iter) -> Self {
230        Self {
231            members: iter.into_iter().collect(),
232        }
233    }
234}
235
236// `Id` doubles as both the admitted `Out` and the rejected `Err` — there is
237// no richer rejection reason than "this id was not a member", so reusing the
238// id itself needs no new error type.
239impl<Id: Ord + Clone + core::fmt::Debug + Send + Sync + 'static> SendPipe for IdSet<Id> {
240    type In = Id;
241    type Out = Id;
242    type Err = Id;
243
244    fn call(&self, id: Id) -> impl Future<Output = Result<Id, Id>> + Send {
245        let admitted = self.contains(&id);
246        async move { if admitted { Ok(id) } else { Err(id) } }
247    }
248}
249
250/// The control-plane update algebra for an [`IdSet`] filter, folded in through
251/// [`FilterControl::apply`]. `Add`/`Remove` grow or shrink the live set
252/// incrementally; `Replace`/`Clear` swap it wholesale. Serializable, so it is
253/// also the wire form of a control message ("add id X" over the network).
254#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
255pub enum FilterUpdate<Id> {
256    Add(Id),
257    Remove(Id),
258    Replace(Vec<Id>),
259    Clear,
260}
261
262// fluent control handle: chaining is the ergonomic point, and the returned
263// `&Self` is optional at each call, so `#[must_use]` would wrongly warn on the
264// common statement form (`control.add(a);`).
265#[allow(clippy::return_self_not_must_use)]
266impl<Id: Ord + Clone> FilterControl<IdSet<Id>> {
267    /// Apply an [`IdSet`] update to the live filter.
268    pub fn apply(&self, update: FilterUpdate<Id>) -> &Self {
269        match update {
270            FilterUpdate::Add(id) => self.update(move |set| set.with(id.clone())),
271            FilterUpdate::Remove(id) => self.update(move |set| set.without(&id)),
272            FilterUpdate::Replace(members) => self.replace(members.into_iter().collect()),
273            FilterUpdate::Clear => self.replace(IdSet::new()),
274        }
275        self
276    }
277
278    /// Add an id to the live set (sugar over [`apply`](Self::apply)).
279    pub fn add(&self, id: Id) -> &Self {
280        self.apply(FilterUpdate::Add(id))
281    }
282
283    /// Remove an id from the live set.
284    pub fn remove(&self, id: Id) -> &Self {
285        self.apply(FilterUpdate::Remove(id))
286    }
287
288    /// Clear the live set (match nothing).
289    pub fn clear(&self) -> &Self {
290        self.apply(FilterUpdate::Clear)
291    }
292
293    /// Replace the live set wholesale from an id iterator.
294    pub fn replace_ids(&self, ids: impl IntoIterator<Item = Id>) -> &Self {
295        self.apply(FilterUpdate::Replace(ids.into_iter().collect()))
296    }
297}
298
299#[cfg(test)]
300#[allow(clippy::unwrap_used, clippy::expect_used, clippy::ptr_arg)]
301mod tests {
302    use super::*;
303
304    // realistic request-correlation ids (hex trace-id prefixes).
305    fn ids() -> (String, String, String) {
306        (
307            "req-7f3a9c2e1b4d".to_string(),
308            "req-0f43d9c0aa18".to_string(),
309            "req-c06db8bb5521".to_string(),
310        )
311    }
312
313    #[test]
314    fn decides_membership_against_the_current_set() {
315        let (subscribed, other, _) = ids();
316        let (filter, _control) = live_filter(IdSet::from_ids([subscribed.clone()]));
317        assert!(filter.contains(&subscribed));
318        assert!(!filter.contains(&other));
319    }
320
321    #[test]
322    fn replace_swaps_the_predicate_seen_by_an_existing_filter() {
323        let (first, second, _) = ids();
324        let (filter, control) = live_filter(IdSet::from_ids([first.clone()]));
325        assert!(filter.contains(&first));
326        control.replace(IdSet::from_ids([second.clone()]));
327        assert!(
328            !filter.contains(&first),
329            "old id no longer matches after replace"
330        );
331        assert!(filter.contains(&second), "new id matches after replace");
332    }
333
334    #[test]
335    fn add_grows_and_remove_shrinks_the_live_set() {
336        let (first, second, third) = ids();
337        let (filter, control) = live_filter(IdSet::from_ids([first.clone()]));
338
339        control.apply(FilterUpdate::Add(second.clone()));
340        assert!(filter.contains(&first));
341        assert!(filter.contains(&second));
342        assert!(!filter.contains(&third));
343
344        control.apply(FilterUpdate::Remove(first.clone()));
345        assert!(!filter.contains(&first), "removed id stops matching");
346        assert!(filter.contains(&second), "untouched id still matches");
347    }
348
349    #[test]
350    fn clear_matches_nothing_then_replace_reseeds() {
351        let (first, second, _) = ids();
352        let (filter, control) = live_filter(IdSet::from_ids([first.clone()]));
353        control.apply(FilterUpdate::Clear);
354        assert!(!filter.contains(&first));
355        assert_eq!(filter.snapshot().len(), 0);
356        control.apply(FilterUpdate::Replace(
357            [second.clone()].into_iter().collect(),
358        ));
359        assert!(filter.contains(&second));
360    }
361
362    #[test]
363    fn all_clones_share_one_live_predicate() {
364        // the split: many subscribers (filter clones), one control — an update
365        // through the control is visible to every subscriber.
366        let (first, second, _) = ids();
367        let (filter, control) = live_filter(IdSet::from_ids([first.clone()]));
368        let subscriber_a = filter.clone();
369        let subscriber_b = filter.clone();
370        control.apply(FilterUpdate::Add(second.clone()));
371        assert!(subscriber_a.contains(&second));
372        assert!(subscriber_b.contains(&second));
373    }
374
375    #[test]
376    fn live_filter_satisfies_the_pipe_seam() {
377        // proves the filtering half is exactly what `.and_then(inner)` wants
378        // (the RISC-reuse claim), at the type level: any `SendPipe<In = Out =
379        // Item>` composes, and `LiveFilter<IdSet<Item>>` is one.
380        fn admits<Predicate>(predicate: &Predicate, id: String) -> bool
381        where
382            Predicate: SendPipe<In = String, Out = String>,
383        {
384            let mut call = core::pin::pin!(SendPipe::call(predicate, id));
385            let waker = core::task::Waker::noop();
386            let mut context = core::task::Context::from_waker(waker);
387            loop {
388                if let core::task::Poll::Ready(outcome) = call.as_mut().poll(&mut context) {
389                    return outcome.is_ok();
390                }
391            }
392        }
393        let (subscribed, other, _) = ids();
394        let (filter, _control) = live_filter(IdSet::from_ids([subscribed.clone()]));
395        assert!(admits(&filter, subscribed));
396        assert!(!admits(&filter, other));
397    }
398
399    #[test]
400    fn update_derives_next_from_current() {
401        // generic update path (not id-set specific): mutate reads the current set.
402        let (first, second, _) = ids();
403        let (filter, control) = live_filter(IdSet::from_ids([first.clone()]));
404        control.update(|set| set.with(second.clone()));
405        assert!(filter.contains(&first));
406        assert!(filter.contains(&second));
407    }
408
409    #[test]
410    fn fluent_control_sugars_chain() {
411        let (first, second, third) = ids();
412        let (filter, control) = live_filter_ids([first.clone()]);
413        control
414            .add(second.clone())
415            .add(third.clone())
416            .remove(first.clone());
417        assert!(!filter.contains(&first), "removed id gone");
418        assert!(filter.contains(&second), "added id present");
419        assert!(filter.contains(&third), "added id present");
420    }
421
422    #[test]
423    fn id_set_config_mirror_round_trips() {
424        let (first, second, _) = ids();
425        let set = IdSet::from_ids([first, second]);
426        let serialized = serde_json::to_string(&set).unwrap();
427        assert!(
428            serialized.starts_with('['),
429            "an id-set is a bare array of ids"
430        );
431        let restored: IdSet<String> = serde_json::from_str(&serialized).unwrap();
432        assert_eq!(restored, set, "config mirror round-trips");
433    }
434
435    #[test]
436    fn filter_update_is_a_serializable_control_message() {
437        let (_, _, third) = ids();
438        let update = FilterUpdate::Add(third);
439        let serialized = serde_json::to_string(&update).unwrap();
440        let restored: FilterUpdate<String> = serde_json::from_str(&serialized).unwrap();
441        assert_eq!(
442            restored, update,
443            "control message round-trips over the wire"
444        );
445    }
446}
proxima_primitives/pipe/live_filter.rs

proxima_primitives/pipe/
live_filter.rs
