liminal_server/console/hub/
taps.rs

1//! Tap channels follow their live subscribers and admission reservations.
2
3use super::{
4    AttachedTap, ConsoleError, ConsoleFailure, ConsoleHub, MAX_LIVE_TAPS, TAP_SUBJECT_PREFIX,
5    TOPOLOGY_SUBJECT, TapError, TapFilter, add, failed, len_u64, lock, pending_total, set,
6};
7use std::collections::VecDeque;
8
9impl ConsoleHub {
10    /// Whether a new subscription to `channel` is within the live-tap limit:
11    /// always for a channel that is not a tap or whose tap is live.
12    ///
13    /// # Errors
14    /// Returns [`TapError::Poisoned`] when the tap table is poisoned.
15    pub fn tap_room(&self, channel: &str) -> Result<bool, TapError> {
16        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
17            return Ok(true);
18        }
19        let taps = lock(&self.taps, "console taps")?;
20        Ok(taps.live.contains_key(channel)
21            || taps.reserved.contains_key(channel)
22            || taps.live.len() + taps.reserved.len() < MAX_LIVE_TAPS)
23    }
24
25    /// Admits a subscribe to `channel` against [`MAX_LIVE_TAPS`] and, for a
26    /// tap not yet live, reserves its slot in the same lock, so two racing
27    /// subscribes cannot both take the last slot. Answers false when no slot
28    /// is left. The attach that follows takes the reservation up;
29    /// [`Self::release_tap`] gives it back when the subscribe fails.
30    ///
31    /// # Errors
32    /// Returns [`TapError::Poisoned`] when the tap table is poisoned.
33    pub fn reserve_tap(&self, channel: &str) -> Result<bool, TapError> {
34        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
35            return Ok(true);
36        }
37        let mut taps = lock(&self.taps, "console taps")?;
38        let admitted = if taps.live.contains_key(channel) {
39            true
40        } else if let Some(holders) = taps.reserved.get_mut(channel) {
41            *holders = holders.saturating_add(1);
42            true
43        } else if taps.live.len() + taps.reserved.len() < MAX_LIVE_TAPS {
44            taps.reserved.insert(channel.to_owned(), 1);
45            true
46        } else {
47            false
48        };
49        drop(taps);
50        Ok(admitted)
51    }
52
53    /// Gives back one reservation of `channel` for a subscribe that failed
54    /// after admission. When it was the last and no tap is live there, the
55    /// tap channel admission registered is taken off the roster by `retire`,
56    /// under the taps lock, so no subscribe reserving the tap again can see
57    /// its entry go: a refused subscribe leaves the roster as it found it. A
58    /// failed `retire` queues the subject for the pump; a poisoned table is
59    /// counted as a detach failure.
60    pub fn release_tap(
61        &self,
62        channel: &str,
63        retire: impl FnOnce(&str) -> Result<(), ConsoleError>,
64    ) {
65        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
66            return;
67        }
68        let Ok(mut taps) = lock(&self.taps, "console taps") else {
69            failed(
70                &self.counts.tap_detach_failures,
71                1,
72                ConsoleFailure::TapDetach,
73            );
74            return;
75        };
76        let last = match taps.reserved.get_mut(channel) {
77            Some(holders) if *holders > 1 => {
78                *holders -= 1;
79                false
80            }
81            Some(_) => {
82                taps.reserved.remove(channel);
83                true
84            }
85            None => false,
86        };
87        if !last || taps.live.contains_key(channel) || retire(channel).is_ok() {
88            return;
89        }
90        taps.retiring.push(channel.to_owned());
91        drop(taps);
92        failed(
93            &self.counts.tap_detach_failures,
94            1,
95            ConsoleFailure::TapDetach,
96        );
97        self.wake();
98    }
99
100    /// Counts a subscription to `channel`. A new `console.topology`
101    /// subscriber makes the latest document due again, so it is sent the
102    /// whole topology (R7); the first to a `console.tap.<target>` starts that
103    /// target's tap. Any other channel is ignored.
104    ///
105    /// # Errors
106    /// Returns [`ConsoleError::TapLimit`] when a new tap would pass
107    /// [`MAX_LIVE_TAPS`], or [`ConsoleError::Tap`] when a lock is poisoned;
108    /// either is counted in `tap_attach_failures`.
109    pub fn attach(&self, channel: &str) -> Result<(), ConsoleError> {
110        if channel == TOPOLOGY_SUBJECT {
111            let mut state = self.lock_topology()?;
112            state.subscribers = state.subscribers.saturating_add(1);
113            state.stale = true;
114            drop(state);
115            self.wake();
116            return Ok(());
117        }
118        let Some(target) = channel.strip_prefix(TAP_SUBJECT_PREFIX) else {
119            return Ok(());
120        };
121        self.attach_tap(channel, target).inspect_err(|_| {
122            failed(
123                &self.counts.tap_attach_failures,
124                1,
125                ConsoleFailure::TapAttach,
126            );
127        })
128    }
129
130    fn attach_tap(&self, channel: &str, target: &str) -> Result<(), ConsoleError> {
131        let mut taps = lock(&self.taps, "console taps")?;
132        match taps.reserved.get_mut(channel) {
133            Some(holders) if *holders > 1 => *holders -= 1,
134            Some(_) => {
135                taps.reserved.remove(channel);
136            }
137            None => {}
138        }
139        if let Some(entry) = taps.live.get_mut(channel) {
140            entry.subscribers = entry.subscribers.saturating_add(1);
141            return Ok(());
142        }
143        if taps.live.len() >= MAX_LIVE_TAPS {
144            return Err(ConsoleError::TapLimit {
145                subject: channel.to_owned(),
146                limit: MAX_LIVE_TAPS,
147            });
148        }
149        let subscription = self.tap.subscribe(TapFilter {
150            subject: target.to_owned(),
151        })?;
152        taps.retiring.retain(|subject| subject != channel);
153        let generation = taps.generations;
154        taps.generations = taps.generations.wrapping_add(1);
155        taps.live.insert(
156            channel.to_owned(),
157            AttachedTap {
158                subscription,
159                subscribers: 1,
160                pending: VecDeque::new(),
161                generation,
162            },
163        );
164        set(&self.counts.taps_attached, len_u64(taps.live.len()));
165        drop(taps);
166        Ok(())
167    }
168
169    /// Uncounts a subscription to `channel`. The last one to leave a
170    /// `console.tap.<target>` stops that target's tap and queues the channel
171    /// for roster removal by the next pump. Safe on the channel actor's
172    /// thread: it takes only hub locks and wakes without waiting.
173    ///
174    /// # Errors
175    /// Returns [`TapError::Poisoned`] when a lock is poisoned; counted in
176    /// `tap_detach_failures`.
177    pub fn detach(&self, channel: &str) -> Result<(), TapError> {
178        self.detach_inner(channel).inspect_err(|_| {
179            failed(
180                &self.counts.tap_detach_failures,
181                1,
182                ConsoleFailure::TapDetach,
183            );
184        })
185    }
186
187    fn detach_inner(&self, channel: &str) -> Result<(), TapError> {
188        if channel == TOPOLOGY_SUBJECT {
189            let mut state = self.lock_topology()?;
190            state.subscribers = state.subscribers.saturating_sub(1);
191            drop(state);
192            return Ok(());
193        }
194        if !channel.starts_with(TAP_SUBJECT_PREFIX) {
195            return Ok(());
196        }
197        let mut taps = lock(&self.taps, "console taps")?;
198        let last = taps.live.get_mut(channel).is_some_and(|entry| {
199            entry.subscribers = entry.subscribers.saturating_sub(1);
200            entry.subscribers == 0
201        });
202        if !last {
203            return Ok(());
204        }
205        // R9: its pending records and unread ring leave with the tap, since
206        // nobody is left to read them; each is counted by name.
207        if let Some(entry) = taps.live.remove(channel) {
208            add(
209                &self.counts.tap_records_abandoned,
210                len_u64(entry.pending.len()),
211            );
212            add(
213                &self.counts.tap_messages_abandoned,
214                len_u64(entry.subscription.held().unwrap_or_default()),
215            );
216        }
217        taps.retiring.push(channel.to_owned());
218        set(&self.counts.taps_attached, len_u64(taps.live.len()));
219        let pending = pending_total(&taps.live);
220        drop(taps);
221        set(&self.counts.console_pending_records, pending);
222        self.wake();
223        Ok(())
224    }
225
226    /// Subjects with a live tap, sorted.
227    ///
228    /// # Errors
229    /// Returns [`TapError::Poisoned`] when the tap table is poisoned.
230    pub fn attached_subjects(&self) -> Result<Vec<String>, TapError> {
231        Ok(lock(&self.taps, "console taps")?
232            .live
233            .keys()
234            .cloned()
235            .collect())
236    }
237}
liminal_server/console/hub/taps.rs

liminal_server/console/hub/
taps.rs
