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

liminal_server/console/
hub.rs
