liminal_server/tap/subscription/
reader.rs

1//! Subscription drains and release preserve their bounded sampling state.
2
3use super::{Admitted, NonZeroU32, Ordering, TapBatch, TapError, TapFilter, TapSubscription, lock};
4
5impl TapSubscription {
6    /// The filter this subscription was made with.
7    #[must_use]
8    pub fn filter(&self) -> &TapFilter {
9        &self.subscriber.filter
10    }
11
12    /// Takes every held message as batches, one per run of messages admitted
13    /// at the same divisor, oldest first, then halves the divisor (the
14    /// subscriber has drained). The evictions since the previous take ride
15    /// the first batch; with no messages and a nonzero eviction count, one
16    /// empty batch carries it. With neither, nothing is returned.
17    ///
18    /// # Errors
19    /// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
20    pub fn take_batches(&self) -> Result<Vec<TapBatch>, TapError> {
21        let mut guard = lock(&self.subscriber.state, "tap subscriber")?;
22        let state = &mut *guard;
23        let (drained, dropped) = state.ring.take();
24        let mut batches: Vec<TapBatch> = Vec::new();
25        for Admitted { one_in, message } in drained {
26            match batches.last_mut() {
27                Some(open) if open.sampled == one_in => open.messages.push(message),
28                _ => batches.push(TapBatch {
29                    seq: 0,
30                    sampled: one_in,
31                    dropped: 0,
32                    messages: vec![message],
33                }),
34            }
35        }
36        match batches.first_mut() {
37            Some(first) => first.dropped = dropped,
38            None if dropped > 0 => batches.push(TapBatch {
39                seq: 0,
40                sampled: state.one_in,
41                dropped,
42                messages: Vec::new(),
43            }),
44            None => {}
45        }
46        for batch in &mut batches {
47            batch.seq = state.next_seq;
48            state.next_seq = state.next_seq.saturating_add(1);
49        }
50        // Halving 1 stays 1: the divisor never reaches zero.
51        state.one_in = NonZeroU32::new(state.one_in.get() / 2).unwrap_or(NonZeroU32::MIN);
52        drop(guard);
53        Ok(batches)
54    }
55
56    /// The current sampling divisor.
57    ///
58    /// # Errors
59    /// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
60    pub fn sample_one_in(&self) -> Result<NonZeroU32, TapError> {
61        Ok(lock(&self.subscriber.state, "tap subscriber")?.one_in)
62    }
63
64    /// Messages held right now (never above the capacity).
65    ///
66    /// # Errors
67    /// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
68    pub fn held(&self) -> Result<usize, TapError> {
69        Ok(lock(&self.subscriber.state, "tap subscriber")?.ring.len())
70    }
71}
72
73impl Drop for TapSubscription {
74    fn drop(&mut self) {
75        // A poisoned list is refused, not recovered: the entry stays, the
76        // failure is counted, and every later `observe` answers the poison.
77        let Ok(mut subscribers) = lock(&self.shared.subscribers, "tap subscribers") else {
78            self.shared
79                .unsubscribe_failures
80                .fetch_add(1, Ordering::Relaxed);
81            crate::metrics::console_failure(
82                crate::console::counts::ConsoleFailure::TapUnsubscribe,
83                1,
84            );
85            tracing::warn!("TAP_UNSUBSCRIBE_FAILED: the tap subscriber list is poisoned");
86            return;
87        };
88        let id = self.subscriber.id;
89        subscribers.retain(|subscriber| subscriber.id != id);
90        self.shared
91            .active
92            .store(subscribers.len(), Ordering::Release);
93    }
94}
liminal_server/tap/subscription/reader.rs

liminal_server/tap/subscription/
reader.rs
