liminal_server/console/
hub.rs

1//! The console hub the connection services own (SRV-012 R4, R6, R7, R9).
2//!
3//! It holds the topology, the tap, one tap subscription per
4//! `console.tap.<target>` channel with at least one subscriber, and the pump
5//! that turns them into records.
6//!
7//! The pump runs on the publisher's thread and never publishes while it holds
8//! a hub lock: it takes what is due under the lock, lets go, publishes, and
9//! puts back what failed. A tap with a pending record is not drained, so its
10//! ring stays the only buffer and ring eviction the only loss, which the next
11//! batch's `dropped` reports. One tap's failure never stops the others: it
12//! is returned once every tap has been drained, published and retired. The
13//! pending records and unread ring of a tap whose last subscriber left go
14//! with it, as do the unsent records of a tap attached anew before they were
15//! put back; nobody is left to read them, so they are counted by name in
16//! `tap_records_abandoned` and `tap_messages_abandoned`, not in
17//! `console_records_lost`.
18
19use std::collections::{BTreeMap, VecDeque};
20use std::sync::{Arc, Mutex, OnceLock};
21mod placements;
22mod pump;
23mod taps;
24#[cfg(test)]
25mod test_support;
26mod topology;
27use std::time::Instant;
28
29use super::counts::{ConsoleCounts, ConsoleCountsSnapshot, ConsoleFailure, add, failed, set};
30use super::error::ConsoleError;
31use super::model::{TopologyChange, TopologyModel};
32use super::publisher::ServerPublisher;
33use super::records::{canonical_bytes, tap_batch_record};
34use super::wall::WallBase;
35use crate::tap::error::lock;
36use crate::tap::{DEFAULT_TAP_CAPACITY, Tap, TapError, TapFilter, TapSubscription, TapWaker};
37
38/// The topology stream's subject.
39pub const TOPOLOGY_SUBJECT: &str = "console.topology";
40/// The prefix of every tap stream's subject; the rest names the target.
41pub const TAP_SUBJECT_PREFIX: &str = "console.tap.";
42/// The most taps live at once (R6); one more is `CONSOLE_TAP_LIMIT_REFUSED`.
43pub const MAX_LIVE_TAPS: usize = 256;
44
45#[derive(Debug)]
46struct TopologyState {
47    model: TopologyModel,
48    stale: bool,
49    subscribers: usize,
50    published: Option<(u64, std::collections::BTreeSet<String>)>,
51}
52
53#[derive(Debug)]
54struct AttachedTap {
55    subscription: TapSubscription,
56    subscribers: usize,
57    pending: VecDeque<Vec<u8>>,
58    /// Which attach of this subject it is, so the pump never puts an earlier
59    /// tap's unsent records back into a later one (SRV-012 R9).
60    generation: u64,
61}
62
63#[derive(Debug, Default)]
64struct Taps {
65    live: BTreeMap<String, AttachedTap>,
66    /// New taps admitted and not yet attached, with how many subscribes hold
67    /// each: they count against [`MAX_LIVE_TAPS`] from admission on.
68    reserved: BTreeMap<String, u64>,
69    retiring: Vec<String>,
70    /// The next tap's [`AttachedTap::generation`].
71    generations: u64,
72}
73
74type NotifierSlot = Arc<Mutex<Option<TapWaker>>>;
75
76/// One of the hub's locks, named for a test that poisons it.
77#[cfg(test)]
78#[derive(Clone, Copy, Debug)]
79pub(crate) enum HubLock {
80    /// The held topology.
81    Topology,
82    /// The live taps.
83    Taps,
84    /// The publisher's notifier slot.
85    Notifier,
86}
87
88/// What one pump did.
89#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
90pub struct PumpReport {
91    /// `console.topology` records published.
92    pub topology_records: u64,
93    /// `console.tap.<target>` records published.
94    pub tap_records: u64,
95    /// Publishes that failed; each record is kept for the next pump.
96    pub publish_failures: u64,
97    /// Tap channels removed from the roster.
98    pub retired: u64,
99}
100
101/// The console's state, owned by the connection services.
102pub struct ConsoleHub {
103    device: String,
104    wall: WallBase,
105    tap: Tap,
106    topology: Mutex<TopologyState>,
107    taps: Mutex<Taps>,
108    notifier: NotifierSlot,
109    counts: Arc<ConsoleCounts>,
110    placements: OnceLock<Arc<super::tap_placement::TapPlacements>>,
111    /// Every record the publisher took, as the encoder made it. Test-only.
112    #[cfg(test)]
113    sent: Mutex<Vec<(String, Vec<u8>)>>,
114    /// How many topology changes reached [`Self::emit`]. Test-only.
115    #[cfg(test)]
116    emits: std::sync::atomic::AtomicU64,
117    /// How many times the topology lock was taken. Test-only.
118    #[cfg(test)]
119    topology_locks: std::sync::atomic::AtomicU64,
120    /// Notified after each applied topology change. Test-only.
121    #[cfg(test)]
122    applied: std::sync::Condvar,
123    /// A tap subject whose next take fails, as a poisoned ring would.
124    /// Test-only.
125    #[cfg(test)]
126    failing_take: Mutex<Option<String>>,
127}
128
129impl std::fmt::Debug for ConsoleHub {
130    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
131        formatter
132            .debug_struct("ConsoleHub")
133            .field("device", &self.device)
134            .field("tap", &self.tap)
135            .finish_non_exhaustive()
136    }
137}
138
139/// A waker that forwards to whatever notifier is installed when it fires. A
140/// poisoned slot is refused, not recovered: the wake is counted as failed.
141fn forwarding_waker(slot: NotifierSlot, counts: Arc<ConsoleCounts>) -> TapWaker {
142    Arc::new(move || match lock(&slot, "console notifier") {
143        Ok(guard) => {
144            let notifier = guard.clone();
145            drop(guard);
146            if let Some(notifier) = notifier {
147                notifier();
148            }
149        }
150        Err(_poisoned) => failed(
151            &counts.console_wake_failures,
152            1,
153            ConsoleFailure::ConsoleWake,
154        ),
155    })
156}
157
158/// Whether the tap hook may observe a publish on `subject`: never one under
159/// `console.` (the console's own records).
160#[must_use]
161pub fn tappable(subject: &str) -> bool {
162    !crate::server::connection::is_console_subject(subject)
163}
164
165impl ConsoleHub {
166    /// A hub for `device` holding `model`, timing records from `wall`, with
167    /// a tap of the default capacity.
168    ///
169    /// # Errors
170    /// Returns [`ConsoleError::EmptyDevice`] when `device` is blank.
171    pub fn new(device: &str, model: TopologyModel, wall: WallBase) -> Result<Self, ConsoleError> {
172        Self::with_capacity(device, model, wall, DEFAULT_TAP_CAPACITY)
173    }
174
175    /// As [`Self::new`], with each tap subscriber's ring holding `capacity`.
176    ///
177    /// # Errors
178    /// Returns [`ConsoleError::EmptyDevice`] when `device` is blank.
179    pub fn with_capacity(
180        device: &str,
181        model: TopologyModel,
182        wall: WallBase,
183        capacity: usize,
184    ) -> Result<Self, ConsoleError> {
185        if device.trim().is_empty() {
186            return Err(ConsoleError::EmptyDevice);
187        }
188        let notifier = NotifierSlot::default();
189        let counts = Arc::new(ConsoleCounts::default());
190        let depot = model.depot();
191        Ok(Self {
192            device: device.to_owned(),
193            wall,
194            tap: Tap::with_waker(
195                device,
196                capacity,
197                forwarding_waker(Arc::clone(&notifier), Arc::clone(&counts)),
198            )
199            .with_depot(depot),
200            topology: Mutex::new(TopologyState {
201                model,
202                stale: false,
203                subscribers: 0,
204                published: None,
205            }),
206            taps: Mutex::new(Taps::default()),
207            notifier,
208            counts,
209            placements: OnceLock::new(),
210            #[cfg(test)]
211            sent: Mutex::new(Vec::new()),
212            #[cfg(test)]
213            emits: std::sync::atomic::AtomicU64::new(0),
214            #[cfg(test)]
215            topology_locks: std::sync::atomic::AtomicU64::new(0),
216            #[cfg(test)]
217            applied: std::sync::Condvar::new(),
218            #[cfg(test)]
219            failing_take: Mutex::new(None),
220        })
221    }
222
223    /// The device this hub publishes for (`liminal.<device>`).
224    #[must_use]
225    pub fn device(&self) -> &str {
226        &self.device
227    }
228
229    /// The tap.
230    #[must_use]
231    pub const fn tap(&self) -> &Tap {
232        &self.tap
233    }
234
235    /// The station serving a registered local responder subject.
236    ///
237    /// # Errors
238    /// The topology lock is poisoned.
239    pub fn serving_station(&self, subject: &str) -> Result<Option<String>, TapError> {
240        Ok(self.lock_topology()?.model.serving_station(subject))
241    }
242
243    /// Every console counter as it stands.
244    #[must_use]
245    pub fn counts(&self) -> ConsoleCountsSnapshot {
246        self.counts
247            .snapshot(self.tap.dropped_total(), self.tap.unsubscribe_failures())
248    }
249
250    /// Publishes one encoded record and answers whether the publisher took
251    /// it; a refusal is one `console_publish` failure.
252    fn send(&self, out: &dyn ServerPublisher, subject: &str, bytes: &[u8]) -> bool {
253        let taken = out.publish_console(subject, bytes).is_ok();
254        if !taken {
255            failed(
256                &self.counts.console_publish_failures,
257                1,
258                ConsoleFailure::ConsolePublish,
259            );
260        }
261        #[cfg(test)]
262        if taken && let Ok(mut sent) = self.sent.lock() {
263            sent.push((subject.to_owned(), bytes.to_vec()));
264        }
265        taken
266    }
267
268    /// The topology lock, counted under test.
269    fn lock_topology(&self) -> Result<std::sync::MutexGuard<'_, TopologyState>, TapError> {
270        #[cfg(test)]
271        self.topology_locks
272            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
273        lock(&self.topology, "console topology")
274    }
275
276    /// This server's incarnation started at `started`; the topology's uptime
277    /// runs from there and the next document says so.
278    ///
279    /// # Errors
280    /// Returns [`TapError::Poisoned`] when the topology lock is poisoned.
281    pub fn incarnation_started(&self, started: Instant) -> Result<(), TapError> {
282        let mut state = self.lock_topology()?;
283        if state.model.incarnation_started(started) {
284            state.stale = true;
285        }
286        drop(state);
287        Ok(())
288    }
289
290    /// Counts a dedup-suppressed re-publish, which is not tapped.
291    pub fn suppressed(&self) {
292        add(&self.counts.tap_suppressed_total, 1);
293    }
294
295    /// Whether the taps lock is held at this moment. Test-only.
296    #[cfg(test)]
297    pub(crate) fn taps_held(&self) -> bool {
298        self.taps.try_lock().is_err()
299    }
300
301    /// How many tap channels hold an admission reservation. Test-only.
302    #[cfg(test)]
303    pub(crate) fn reserved_taps(&self) -> Option<usize> {
304        self.taps.lock().ok().map(|taps| taps.reserved.len())
305    }
306
307    /// Every record the publisher took, oldest first, as the encoder made
308    /// it; `None` when the capture is poisoned. Test-only.
309    #[cfg(test)]
310    pub(crate) fn sent(&self) -> Option<Vec<(String, Vec<u8>)>> {
311        self.sent.lock().ok().map(|sent| sent.clone())
312    }
313
314    /// Poisons one of the hub's locks, as a panic on a thread holding it
315    /// would, and answers whether it is now poisoned. Test-only.
316    #[cfg(test)]
317    pub(crate) fn poison(&self, lock: HubLock) -> bool {
318        std::thread::scope(|scope| {
319            scope
320                .spawn(|| {
321                    let _held = (
322                        matches!(lock, HubLock::Topology).then(|| self.topology.lock()),
323                        matches!(lock, HubLock::Taps).then(|| self.taps.lock()),
324                        matches!(lock, HubLock::Notifier).then(|| self.notifier.lock()),
325                    );
326                    std::panic::resume_unwind(Box::new("poisoned for a test"));
327                })
328                .join()
329                .is_err()
330        }) && match lock {
331            HubLock::Topology => self.topology.is_poisoned(),
332            HubLock::Taps => self.taps.is_poisoned(),
333            HubLock::Notifier => self.notifier.is_poisoned(),
334        }
335    }
336
337    /// Counts a publish the tap could not observe (its lock was poisoned).
338    /// The publish itself answered as it would without the console.
339    pub fn observe_failed(&self) {
340        failed(
341            &self.counts.tap_observe_failures,
342            1,
343            ConsoleFailure::TapObserve,
344        );
345    }
346
347    /// Installs the function that tells the publisher there is work. The
348    /// publisher calls [`Self::pump`] on its own thread when told.
349    ///
350    /// # Errors
351    /// Returns [`TapError::Poisoned`] when the slot is poisoned.
352    pub fn set_notifier(&self, notifier: TapWaker) -> Result<(), TapError> {
353        *lock(&self.notifier, "console notifier")? = Some(notifier);
354        Ok(())
355    }
356
357    fn wake(&self) {
358        forwarding_waker(Arc::clone(&self.notifier), Arc::clone(&self.counts))();
359    }
360
361    /// Wakes the publisher when work is already waiting: a topology
362    /// subscriber, a live tap or a tap to retire. The start of the console
363    /// streams calls it, so a subscriber that arrived before the publisher
364    /// ran is served, and an idle server is never woken.
365    ///
366    /// # Errors
367    /// Returns [`TapError::Poisoned`] when a hub lock is poisoned.
368    pub fn wake_if_due(&self) -> Result<(), TapError> {
369        let topology = self.lock_topology()?.subscribers > 0;
370        let taps = {
371            let taps = lock(&self.taps, "console taps")?;
372            !taps.live.is_empty() || !taps.retiring.is_empty()
373        };
374        if topology || taps {
375            self.wake();
376        }
377        Ok(())
378    }
379
380    /// Applies `change` to the held topology. A change that alters it makes
381    /// the document due, and wakes the publisher when someone subscribes.
382    ///
383    /// # Errors
384    /// Returns [`TapError::Poisoned`] when the topology lock is poisoned; the
385    /// change is counted in `topology_emit_failures`.
386    pub fn emit(&self, change: TopologyChange) -> Result<(), TapError> {
387        #[cfg(test)]
388        self.emits
389            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
390        let emit_failed = |_: &TapError| {
391            failed(
392                &self.counts.topology_emit_failures,
393                1,
394                ConsoleFailure::TopologyEmit,
395            );
396        };
397        // A worker identity or service subject that is this server's exchange
398        // or depot id is kept off the map and counted (Buttercup's V1, Hermes
399        // 04:27): the registration stands, merged into nothing.
400        let colliding = match &change {
401            TopologyChange::Worker { identity, .. } => Some(identity),
402            TopologyChange::Service {
403                subject,
404                present: true,
405            } => Some(subject),
406            _ => None,
407        }
408        .and_then(|id| self.tap.reserved(id).map(|station| (id, station)));
409        if let Some((id, station)) = colliding {
410            failed(
411                &self.counts.station_collisions,
412                1,
413                ConsoleFailure::StationCollision,
414            );
415            tracing::warn!(id = %id, station, "STATION_COLLISION");
416        }
417        if colliding.is_none() {
418            self.offer_change(&change);
419        }
420        // A registered worker's tapped publishes name its station from here
421        // to its connection's close.
422        match &change {
423            TopologyChange::Worker {
424                connection,
425                identity,
426                ..
427            } if colliding.is_none() => self.tap.name_station(*connection, identity),
428            TopologyChange::ConnectionClosed { id } => self
429                .tap
430                .forget_station(*id)
431                .and_then(|()| self.tap.forget_receivers_of(*id)),
432            _ => Ok(()),
433        }
434        .inspect_err(emit_failed)?;
435        let mut state = self.lock_topology().inspect_err(emit_failed)?;
436        let changed = state.model.apply(change);
437        if changed {
438            state.stale = true;
439        }
440        let due = changed && state.subscribers > 0;
441        drop(state);
442        #[cfg(test)]
443        self.applied.notify_all();
444        if due {
445            self.wake();
446        }
447        Ok(())
448    }
449
450    /// Publishes what is due through `out`: the topology document when it
451    /// changed and someone subscribes, then each live tap's records, a
452    /// tap's kept records first and its ring drained only when none are
453    /// kept; then removes retired tap channels from the roster. A failed
454    /// publish keeps that record and the tap's later ones for the next pump,
455    /// and the pump goes on to the other taps.
456    ///
457    /// # Errors
458    /// Returns [`ConsoleError`] when a lock is poisoned or a record cannot be
459    /// encoded; publish failures are counted and reported, not returned.
460    pub fn pump(&self, out: &dyn ServerPublisher) -> Result<PumpReport, ConsoleError> {
461        add(&self.counts.console_pump_wakes, 1);
462        let mut report = PumpReport::default();
463        // The taps pump whatever the topology did, so one failure there
464        // never holds the feed.
465        let topology = self.pump_topology(out, &mut report);
466        let tapped = self.pump_taps(out, &mut report);
467        self.retire(out, &mut report)?;
468        topology.and(tapped).map(|()| report)
469    }
470}
471
472fn len_u64(len: usize) -> u64 {
473    u64::try_from(len).unwrap_or(u64::MAX)
474}
475
476fn pending_total(live: &BTreeMap<String, AttachedTap>) -> u64 {
477    live.values()
478        .map(|entry| len_u64(entry.pending.len()))
479        .sum()
480}
liminal_server/console/hub.rs

liminal_server/console/
hub.rs
