liminal-server 0.21.1

//! The tap and its subscriptions.

use liminal::channel::{PublishRecord, Recipient};
use std::collections::HashMap;
use std::num::NonZeroU32;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Instant;

use super::error::{TapError, lock};
use super::message::{
    Hop, PayloadGrant, PublishFact, Reached, TapBatch, TapFilter, TapIdent, TapMessage, TapSource,
};
use super::ring::BoundedRing;

/// Default per-subscriber ring capacity, in messages.
pub const DEFAULT_TAP_CAPACITY: usize = 4096;

/// The largest sampling divisor the server will choose.
pub const MAX_SAMPLE_ONE_IN: NonZeroU32 = match NonZeroU32::new(1 << 16) {
    Some(most) => most,
    None => NonZeroU32::MIN,
};

const TWO: NonZeroU32 = NonZeroU32::MIN.saturating_add(1);

/// Wakes whoever drains the tap; called at most once per publish, after every
/// tap lock is released.
pub type TapWaker = Arc<dyn Fn() + Send + Sync>;

/// One kept message and the divisor it was admitted under, so a drain can
/// close a batch wherever the divisor changed without a second ring.
#[derive(Debug)]
struct Admitted {
    one_in: NonZeroU32,
    message: TapMessage,
}

#[derive(Debug)]
struct SubscriberState {
    ring: BoundedRing<Admitted>,
    one_in: NonZeroU32,
    seen: u64,
    next_seq: u64,
}

impl SubscriberState {
    /// The sampling decision for one matching publish, taken before anything
    /// is allocated for it.
    fn admits(&mut self) -> bool {
        let take = self.seen % u64::from(self.one_in.get()) == 0;
        self.seen = self.seen.wrapping_add(1);
        take
    }

    /// Doubles the divisor once the ring is more than half full.
    fn after_push(&mut self) {
        if self.ring.len().saturating_mul(2) > self.ring.capacity() {
            self.one_in = self.one_in.saturating_mul(TWO).min(MAX_SAMPLE_ONE_IN);
        }
    }
}

struct Subscriber {
    id: u64,
    filter: TapFilter,
    grant: Option<PayloadGrant>,
    state: Mutex<SubscriberState>,
}

impl std::fmt::Debug for Subscriber {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("Subscriber")
            .field("id", &self.id)
            .field("filter", &self.filter)
            .finish_non_exhaustive()
    }
}

#[derive(Default)]
struct TapShared {
    active: AtomicUsize,
    observed: AtomicU64,
    /// Of `observed`, the server's own publishes (`liminal.<device>`).
    /// Test-only.
    #[cfg(test)]
    observed_server: AtomicU64,
    dropped_total: AtomicU64,
    unsubscribe_failures: AtomicU64,
    next_id: AtomicU64,
    subscribers: Mutex<Vec<Arc<Subscriber>>>,
    /// Each registered worker's connection and its station id, so a kept
    /// message names its publishing station without the topology lock.
    stations: Mutex<HashMap<u64, Arc<str>>>,
    /// Each live subscription's subscriber pid and its connection, so a
    /// recipient the fan-out recorded names its connection's station.
    receivers: Mutex<HashMap<u64, u64>>,
    /// Recipient buffers between publishes, so a recording publish reuses
    /// an allocation rather than making one.
    buffers: Mutex<Vec<PublishRecord>>,
    /// Recipient buffers lent: publishes that recorded their recipients.
    lent: AtomicU64,
    /// The one empty station list, shared by every kept message whose
    /// recorded fan-out reached no registered worker, so naming none
    /// allocates nothing.
    no_stations: Arc<[Reached]>,
    waker: OnceLock<TapWaker>,
}

impl std::fmt::Debug for TapShared {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("TapShared")
            .field("active", &self.active)
            .field("observed", &self.observed)
            .field("dropped_total", &self.dropped_total)
            .finish_non_exhaustive()
    }
}

/// The server's tap. Cheap to clone; every clone is the same tap.
#[derive(Clone, Debug)]
pub struct Tap {
    shared: Arc<TapShared>,
    node: Arc<str>,
    /// This server's depot's station id, when it keeps a persistent store:
    /// a durable message's hop between its publisher and the exchange.
    depot: Option<Arc<str>>,
    capacity: usize,
}

impl Tap {
    /// A tap whose hops name `node`, with per-subscriber rings of `capacity`,
    /// that wakes nobody: for a reader that drains when it chooses, as a test
    /// does.
    #[must_use]
    pub fn new(node: &str, capacity: usize) -> Self {
        Self::build(node, capacity, OnceLock::new())
    }

    /// A tap that calls `waker` once per publish any subscription kept.
    /// Fixed at construction, so the hot path reads it without a lock and
    /// nothing can replace it.
    #[must_use]
    pub fn with_waker(node: &str, capacity: usize, waker: TapWaker) -> Self {
        Self::build(node, capacity, OnceLock::from(waker))
    }

    fn build(node: &str, capacity: usize, waker: OnceLock<TapWaker>) -> Self {
        Self {
            shared: Arc::new(TapShared {
                waker,
                ..TapShared::default()
            }),
            node: Arc::from(node),
            depot: None,
            capacity: capacity.max(1),
        }
    }

    /// This tap, on a server that keeps a persistent store when `depot`:
    /// its kept durable messages then pass `<node>.depot` (lane brief
    /// ruling 8).
    #[must_use]
    pub fn with_depot(mut self, depot: bool) -> Self {
        self.depot = depot.then(|| Arc::from(format!("{}.depot", self.node)));
        self
    }

    /// The station `id` would collide with: this tap's exchange (its node)
    /// or its depot. A worker with such an identity names no station here.
    #[must_use]
    pub fn reserved(&self, id: &str) -> Option<&'static str> {
        if id == &*self.node {
            Some("exchange")
        } else if self.depot.as_deref() == Some(id) {
            Some("depot")
        } else {
            None
        }
    }

    /// Subscribes without payloads.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the subscriber list is poisoned.
    pub fn subscribe(&self, filter: TapFilter) -> Result<TapSubscription, TapError> {
        self.subscribe_with(filter, None)
    }

    /// Subscribes, with payloads when `grant` is present.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the subscriber list is poisoned.
    pub fn subscribe_with(
        &self,
        filter: TapFilter,
        grant: Option<PayloadGrant>,
    ) -> Result<TapSubscription, TapError> {
        let id = self.shared.next_id.fetch_add(1, Ordering::Relaxed);
        let subscriber = Arc::new(Subscriber {
            id,
            filter,
            grant,
            state: Mutex::new(SubscriberState {
                ring: BoundedRing::new(self.capacity),
                one_in: NonZeroU32::MIN,
                seen: 0,
                next_seq: 1,
            }),
        });
        let mut subscribers = lock(&self.shared.subscribers, "tap subscribers")?;
        subscribers.push(Arc::clone(&subscriber));
        self.shared
            .active
            .store(subscribers.len(), Ordering::Release);
        drop(subscribers);
        Ok(TapSubscription {
            shared: Arc::clone(&self.shared),
            subscriber,
        })
    }

    /// Publishes the hook has shown the tap while a subscription was live (a
    /// count pin: N ordinary publishes give N, the tap's own records give
    /// zero).
    #[must_use]
    pub fn observations(&self) -> u64 {
        self.shared.observed.load(Ordering::Relaxed)
    }

    /// Of [`Self::observations`], those of the server's own publishes (its
    /// `liminal.events` and `liminal.cost` rows), which its publisher owner
    /// writes on its own schedule. Test-only.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn server_observations(&self) -> u64 {
        self.shared.observed_server.load(Ordering::Relaxed)
    }

    /// Live subscriptions.
    #[must_use]
    pub fn subscriber_count(&self) -> usize {
        self.shared.active.load(Ordering::Acquire)
    }

    /// Messages evicted from every subscription's ring since the tap was made.
    #[must_use]
    pub fn dropped_total(&self) -> u64 {
        self.shared.dropped_total.load(Ordering::Relaxed)
    }

    /// Poisons the subscriber list, as a panic on a thread holding it would,
    /// and answers whether it is now poisoned. Test-only.
    #[cfg(test)]
    pub(crate) fn poison_subscribers(&self) -> bool {
        std::thread::scope(|scope| {
            scope
                .spawn(|| {
                    let _held = self.shared.subscribers.lock();
                    std::panic::resume_unwind(Box::new("poisoned for a test"));
                })
                .join()
                .is_err()
        }) && self.shared.subscribers.is_poisoned()
    }

    /// Names `station` as the station `connection` publishes as, from its
    /// worker registration until [`Self::forget_station`].
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the station table is poisoned.
    pub fn name_station(&self, connection: u64, station: &str) -> Result<(), TapError> {
        lock(&self.shared.stations, "tap stations")?.insert(connection, Arc::from(station));
        Ok(())
    }

    /// Forgets `connection`'s station, as its connection closes.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the station table is poisoned.
    pub fn forget_station(&self, connection: u64) -> Result<(), TapError> {
        lock(&self.shared.stations, "tap stations")?.remove(&connection);
        Ok(())
    }

    /// Names `connection` as the owner of the subscription whose subscriber
    /// pid is `subscriber`, until [`Self::forget_receiver`].
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the receiver table is poisoned.
    pub fn name_receiver(&self, subscriber: u64, connection: u64) -> Result<(), TapError> {
        lock(&self.shared.receivers, "tap receivers")?.insert(subscriber, connection);
        Ok(())
    }

    /// Forgets the subscription whose subscriber pid is `subscriber`, as it
    /// leaves its channel.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the receiver table is poisoned.
    pub fn forget_receiver(&self, subscriber: u64) -> Result<(), TapError> {
        lock(&self.shared.receivers, "tap receivers")?.remove(&subscriber);
        Ok(())
    }

    /// Forgets every subscription of `connection`, as it closes.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the receiver table is poisoned.
    pub fn forget_receivers_of(&self, connection: u64) -> Result<(), TapError> {
        lock(&self.shared.receivers, "tap receivers")?.retain(|_, owner| *owner != connection);
        Ok(())
    }

    /// A buffer for a publish to record its recipients in, only while a
    /// subscription is live: with none it is one atomic load, no lock and no
    /// allocation, and the publish records nothing. A buffer lent comes from
    /// those given back, so it allocates only while the pool and each buffer
    /// grow to the widest concurrent fan-out.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the buffer pool is poisoned.
    pub fn lend_recipients(&self) -> Result<Option<PublishRecord>, TapError> {
        if self.shared.active.load(Ordering::Acquire) == 0 {
            return Ok(None);
        }
        self.shared.lent.fetch_add(1, Ordering::Relaxed);
        Ok(Some(
            lock(&self.shared.buffers, "tap buffers")?
                .pop()
                .unwrap_or_default(),
        ))
    }

    /// Takes back a buffer [`Self::lend_recipients`] lent, emptied, for the
    /// next recording publish.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the buffer pool is poisoned.
    pub fn give_back(&self, mut buffer: PublishRecord) -> Result<(), TapError> {
        buffer.clear();
        lock(&self.shared.buffers, "tap buffers")?.push(buffer);
        Ok(())
    }

    /// Recipient buffers lent since the tap was made: publishes that asked
    /// the fan-out for their recipients.
    #[must_use]
    pub fn recipients_lent(&self) -> u64 {
        self.shared.lent.load(Ordering::Relaxed)
    }

    /// Subscriptions whose drop found the subscriber list poisoned and so
    /// could not leave it.
    #[must_use]
    pub fn unsubscribe_failures(&self) -> u64 {
        self.shared.unsubscribe_failures.load(Ordering::Relaxed)
    }

    /// The hook: one publish return. With no subscription it is one atomic
    /// load and nothing else. Otherwise each matching subscription decides
    /// under its own lock whether to keep the message before anything is
    /// allocated for it; a kept message allocates its subject and its hops
    /// (and its reply-to, and a payload copy under a grant), and the drainer
    /// is woken once, after every tap lock is released. Answers whether any
    /// subscription kept it.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when a tap lock is poisoned; any
    /// subscription that kept the message before it still wakes the drainer.
    /// Returns [`TapError::StationsDisagree`] when its station list was
    /// refused: the message is kept, its receivers not recorded. Returns
    /// [`TapError::HopsOutOfOrder`] when a hop was measured before the one
    /// ahead of it: the message is kept with its hops as measured. Returns
    /// [`TapError::HopBeforeFirstSight`] or [`TapError::HopOverflow`] when a
    /// hop could not be carried: the message is kept without it. More than
    /// one of these is [`TapError::Several`], each named.
    pub fn observe(&self, fact: &PublishFact<'_>) -> Result<bool, TapError> {
        self.observe_from(fact, None)
    }

    pub(crate) fn observe_embedded(
        &self,
        fact: &PublishFact<'_>,
        host: &str,
    ) -> Result<bool, TapError> {
        self.observe_from(fact, Some(host))
    }

    fn observe_from(&self, fact: &PublishFact<'_>, host: Option<&str>) -> Result<bool, TapError> {
        if self.shared.active.load(Ordering::Acquire) == 0 {
            return Ok(false);
        }
        let observation = self.shared.observed.fetch_add(1, Ordering::Relaxed);
        #[cfg(test)]
        if fact.source == TapSource::Server && host.is_none() {
            self.shared.observed_server.fetch_add(1, Ordering::Relaxed);
        }
        let mut kept = false;
        let result = self.keep(fact, observation, &mut kept, host);
        if kept && let Some(waker) = self.shared.waker.get() {
            waker();
        }
        result.map(|()| kept)
    }

    /// The station `source` publishes as, when it is a registered worker.
    fn station_of(&self, source: TapSource) -> Result<Option<Arc<str>>, TapError> {
        Ok(match source {
            TapSource::Connection(connection) => lock(&self.shared.stations, "tap stations")?
                .get(&connection)
                .map(Arc::clone),
            TapSource::Server => None,
        })
    }

    /// The stations `recipients` belong to: each registered worker once, in
    /// order, with the instant its first recipient was queued, in one
    /// allocation; the shared empty list when none is one, which
    /// allocates nothing. A first pass counts them; the second fills a slice
    /// of exactly that length, so the shared list is built in place.
    fn stations_of(&self, recipients: &[Recipient]) -> Result<Arc<[Reached]>, TapError> {
        if recipients.is_empty() {
            return Ok(Arc::clone(&self.shared.no_stations));
        }
        let receivers = lock(&self.shared.receivers, "tap receivers")?;
        let stations = lock(&self.shared.stations, "tap stations")?;
        let station_at = |index: usize| {
            recipients
                .get(index)
                .and_then(|(subscriber, _)| receivers.get(subscriber))
                .and_then(|connection| stations.get(connection))
        };
        // The recipient at `index` names a station no earlier one named, and
        // was queued at the instant it carries.
        let first_at = |index: usize| {
            station_at(index)
                .filter(|station| !(0..index).any(|earlier| station_at(earlier) == Some(*station)))
                .zip(recipients.get(index).map(|(_, queued_at)| *queued_at))
        };
        let named = (0..recipients.len())
            .filter(|index| first_at(*index).is_some())
            .count();
        let list = gather(recipients.len(), named, first_at)?
            .unwrap_or_else(|| Arc::clone(&self.shared.no_stations));
        drop(stations);
        drop(receivers);
        Ok(list)
    }

    /// A kept message's hops, in one allocation of no more than their
    /// number (lane brief rulings 3 and 8): its publishing station at 0;
    /// this server's depot as a durable append completed, when it keeps a
    /// persistent store; the exchange (this node) once the message was
    /// appended, so an append to an in-memory store sits inside its time;
    /// then each station it reached as its first inbox queued it. No exchange
    /// hop: nothing measured the fan-out. A publish that appended nothing
    /// took no exchange instant, which is normal and not a failure (Jules's
    /// F1, Hermes 07:22), so its exchange is left out, never written at 0.
    /// Each is measured from `seen_at` and kept as measured, never rewritten
    /// (Hermes 04:09 and 04:27). Into `failures`, when given, go the first
    /// hop measured before first sight and the first beyond u64
    /// milliseconds, each left out of the hops, and the first hop measured
    /// before the hop kept ahead of it, kept; each names its place in the
    /// chain.
    fn hops(
        &self,
        fact: &PublishFact<'_>,
        seen_at: Instant,
        station: Option<&Arc<str>>,
        reached: Option<&[Reached]>,
        failures: Option<&mut Vec<TapError>>,
    ) -> Vec<Hop> {
        let reached = reached.unwrap_or_default();
        let appended_at = fact.recorded.and_then(|recorded| recorded.appended_at);
        let depot = self.depot.as_ref().zip(appended_at);
        let mut hops = Vec::with_capacity(
            usize::from(station.is_some())
                + usize::from(depot.is_some())
                + usize::from(appended_at.is_some())
                + reached.len(),
        );
        let mut early = None;
        let mut overflow = None;
        let mut out_of_order = None;
        let mut place = 0;
        let mut pass = |node: &Arc<str>, at: Option<Instant>| {
            let hop = place;
            place += 1;
            let at_ms = match at {
                None => 0,
                Some(at) if at < seen_at => {
                    early.get_or_insert_with(|| {
                        elapsed_ms(seen_at.duration_since(at))
                            .map_or(TapError::HopOverflow { hop }, |early_ms| {
                                TapError::HopBeforeFirstSight { hop, early_ms }
                            })
                    });
                    return;
                }
                Some(at) => {
                    let Some(at_ms) = elapsed_ms(at.duration_since(seen_at)) else {
                        overflow.get_or_insert(TapError::HopOverflow { hop });
                        return;
                    };
                    at_ms
                }
            };
            if let Some(before) = hops.last().map(|kept: &Hop| kept.at_ms)
                && at_ms < before
            {
                out_of_order.get_or_insert(TapError::HopsOutOfOrder {
                    hop,
                    at_ms,
                    before_ms: before,
                });
            }
            hops.push(Hop {
                node: Arc::clone(node),
                at_ms,
            });
        };
        if let Some(station) = station {
            pass(station, None);
        }
        if let Some((depot, appended_at)) = depot {
            pass(depot, Some(appended_at));
        }
        if let Some(appended_at) = appended_at {
            pass(&self.node, Some(appended_at));
        }
        for reached in reached {
            pass(&reached.station, Some(reached.queued_at));
        }
        if let Some(failures) = failures {
            failures.extend([early, overflow, out_of_order].into_iter().flatten());
        }
        hops
    }

    fn keep(
        &self,
        fact: &PublishFact<'_>,
        observation: u64,
        kept: &mut bool,
        host: Option<&str>,
    ) -> Result<(), TapError> {
        let mut seen_at = None;
        // Looked up once, and only when a subscription keeps the message.
        let mut station = None;
        let mut to_stations = None;
        // A station list refused, and hops that could not be kept as
        // measured, are each reported once the message is kept (Buttercup's
        // V3): empty, this allocates nothing.
        let mut failures = Vec::new();
        let mut hops_reported = false;
        let subscribers = lock(&self.shared.subscribers, "tap subscribers")?;
        for subscriber in subscribers.iter() {
            if !subscriber.filter.matches(fact.subject) {
                continue;
            }
            let mut state = lock(&subscriber.state, "tap subscriber")?;
            if !state.admits() {
                continue;
            }
            let one_in = state.one_in;
            if station.is_none() {
                station = Some(match host {
                    Some(host) => Some(Arc::from(format!("liminal.{host}"))),
                    None => self.station_of(fact.source)?,
                });
                to_stations = Some(
                    match fact
                        .recorded
                        .map(|recorded| self.stations_of(recorded.recipients))
                    {
                        Some(Ok(list)) => Some(list),
                        // Kept as not recorded, and the refusal reported.
                        Some(Err(error @ TapError::StationsDisagree { .. })) => {
                            failures.push(error);
                            None
                        }
                        Some(Err(error)) => return Err(error),
                        None => None,
                    },
                );
            }
            // First sight is receipt (lane brief ruling 6), or now when the
            // publish recorded nothing.
            let first_sight = *seen_at.get_or_insert_with(|| {
                fact.recorded
                    .map_or_else(Instant::now, |recorded| recorded.received_at)
            });
            // Every keeping subscription's hops are the same; their failures
            // are taken once.
            let hops = self.hops(
                fact,
                first_sight,
                station.as_ref().and_then(Option::as_ref),
                to_stations.as_ref().and_then(Option::as_deref),
                (!hops_reported).then_some(&mut failures),
            );
            hops_reported = true;
            let message = TapMessage {
                ident: fact
                    .message_id
                    .map_or(TapIdent::Observation(observation), TapIdent::Message),
                subject: fact.subject.to_owned(),
                size: u64::try_from(fact.payload.len()).unwrap_or(u64::MAX),
                outcome: fact.outcome,
                seen_at: first_sight,
                source: fact.source,
                station: station.clone().flatten(),
                to_stations: to_stations.clone().flatten(),
                hops,
                reply_to: fact.reply_to.map(str::to_owned),
                held: false,
                payload: subscriber.grant.map(|_| fact.payload.to_vec()),
            };
            if state.ring.push(Admitted { one_in, message }) {
                self.shared.dropped_total.fetch_add(1, Ordering::Relaxed);
            }
            state.after_push();
            drop(state);
            *kept = true;
        }
        drop(subscribers);
        match failures.len() {
            0 => Ok(()),
            1 => failures.pop().map_or(Ok(()), Err),
            _ => Err(TapError::Several(failures)),
        }
    }
}

/// One subscriber's handle. Dropping it unsubscribes.
#[derive(Debug)]
pub struct TapSubscription {
    shared: Arc<TapShared>,
    subscriber: Arc<Subscriber>,
}

impl TapSubscription {
    /// The filter this subscription was made with.
    #[must_use]
    pub fn filter(&self) -> &TapFilter {
        &self.subscriber.filter
    }

    /// Takes every held message as batches, one per run of messages admitted
    /// at the same divisor, oldest first, then halves the divisor (the
    /// subscriber has drained). The evictions since the previous take ride
    /// the first batch; with no messages and a nonzero eviction count, one
    /// empty batch carries it. With neither, nothing is returned.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
    pub fn take_batches(&self) -> Result<Vec<TapBatch>, TapError> {
        let mut guard = lock(&self.subscriber.state, "tap subscriber")?;
        let state = &mut *guard;
        let (drained, dropped) = state.ring.take();
        let mut batches: Vec<TapBatch> = Vec::new();
        for Admitted { one_in, message } in drained {
            match batches.last_mut() {
                Some(open) if open.sampled == one_in => open.messages.push(message),
                _ => batches.push(TapBatch {
                    seq: 0,
                    sampled: one_in,
                    dropped: 0,
                    messages: vec![message],
                }),
            }
        }
        match batches.first_mut() {
            Some(first) => first.dropped = dropped,
            None if dropped > 0 => batches.push(TapBatch {
                seq: 0,
                sampled: state.one_in,
                dropped,
                messages: Vec::new(),
            }),
            None => {}
        }
        for batch in &mut batches {
            batch.seq = state.next_seq;
            state.next_seq = state.next_seq.saturating_add(1);
        }
        // Halving 1 stays 1: the divisor never reaches zero.
        state.one_in = NonZeroU32::new(state.one_in.get() / 2).unwrap_or(NonZeroU32::MIN);
        drop(guard);
        Ok(batches)
    }

    /// The current sampling divisor.
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
    pub fn sample_one_in(&self) -> Result<NonZeroU32, TapError> {
        Ok(lock(&self.subscriber.state, "tap subscriber")?.one_in)
    }

    /// Messages held right now (never above the capacity).
    ///
    /// # Errors
    /// Returns [`TapError::Poisoned`] when the subscriber state is poisoned.
    pub fn held(&self) -> Result<usize, TapError> {
        Ok(lock(&self.subscriber.state, "tap subscriber")?.ring.len())
    }
}

impl Drop for TapSubscription {
    fn drop(&mut self) {
        // A poisoned list is refused, not recovered: the entry stays, the
        // failure is counted, and every later `observe` answers the poison.
        let Ok(mut subscribers) = lock(&self.shared.subscribers, "tap subscribers") else {
            self.shared
                .unsubscribe_failures
                .fetch_add(1, Ordering::Relaxed);
            crate::metrics::console_failure(
                crate::console::counts::ConsoleFailure::TapUnsubscribe,
                1,
            );
            tracing::warn!("TAP_UNSUBSCRIBE_FAILED: the tap subscriber list is poisoned");
            return;
        };
        let id = self.subscriber.id;
        subscribers.retain(|subscriber| subscriber.id != id);
        self.shared
            .active
            .store(subscribers.len(), Ordering::Release);
    }
}

/// `elapsed` in whole milliseconds; `None` beyond what a u64 holds, which
/// the caller names rather than writing the most a u64 holds.
fn elapsed_ms(elapsed: std::time::Duration) -> Option<u64> {
    u64::try_from(elapsed.as_millis()).ok()
}

/// The `named` first-named stations of `recipients` recipients, in order, in
/// one allocation: `first_at(index)` is the station the recipient at `index`
/// names first, if it names one, with the instant that recipient was queued. `None` when none is named.
///
/// # Errors
/// Returns [`TapError::StationsDisagree`] when the fill finds other than
/// `named` stations.
fn gather<'station>(
    recipients: usize,
    named: usize,
    first_at: impl Fn(usize) -> Option<(&'station Arc<str>, Instant)>,
) -> Result<Option<Arc<[Reached]>>, TapError> {
    let Some(first) = (0..recipients).find_map(&first_at) else {
        return if named == 0 {
            Ok(None)
        } else {
            Err(TapError::StationsDisagree { named, found: 0 })
        };
    };
    // A range mapped is of known length, so the collect allocates once.
    // Each step finds the next first-named station. A step that finds none
    // fills its slot with `first` only so the slice is whole; the count of
    // those found then refuses the list, so that slot never leaves here.
    let mut next = 0;
    let mut found = 0_usize;
    let reached = |(station, queued_at): (&Arc<str>, Instant)| Reached {
        station: Arc::clone(station),
        queued_at,
    };
    let list: Arc<[Reached]> = (0..named)
        .map(|_| {
            match (next..recipients)
                .find_map(|index| first_at(index).map(|station| (index, station)))
            {
                Some((index, station)) => {
                    next = index + 1;
                    found += 1;
                    reached(station)
                }
                None => reached(first),
            }
        })
        .collect();
    let beyond = (next..recipients)
        .filter(|index| first_at(*index).is_some())
        .count();
    if found != named || beyond != 0 {
        return Err(TapError::StationsDisagree {
            named,
            found: found + beyond,
        });
    }
    Ok(Some(list))
}

#[cfg(test)]
mod gather_tests {
    use std::sync::Arc;
    use std::time::{Duration, Instant};

    use super::{Reached, TapError, elapsed_ms, gather};

    /// Buttercup's V2, Hermes 04:27: a hop further out than u64
    /// milliseconds is named, never written as the most a u64 holds.
    #[test]
    fn an_elapsed_time_beyond_u64_milliseconds_has_no_reading() {
        assert_eq!(elapsed_ms(Duration::from_millis(7)), Some(7));
        assert_eq!(elapsed_ms(Duration::MAX), None);
    }

    /// Buttercup's finding 2 (Hermes 03:05): when the filling pass finds
    /// fewer stations than the count named, the list is refused by name,
    /// never filled in with a station that stands in for the missing one.
    /// The count named two; the fill finds only the ledger, as if the audit
    /// worker had left between the passes.
    #[test]
    fn a_fill_that_disagrees_with_the_count_is_refused() {
        let ledger: Arc<str> = Arc::from("orders.ledger");
        let queued = Instant::now();
        let gathered = gather(2, 2, |index| (index == 0).then_some((&ledger, queued)));
        assert!(
            matches!(
                gathered,
                Err(TapError::StationsDisagree { named: 2, found: 1 })
            ),
            "{gathered:?}"
        );
    }

    /// Positive control: passes that agree build the list, and none named
    /// builds none.
    #[test]
    fn a_fill_that_agrees_builds_the_list() -> Result<(), TapError> {
        let ledger: Arc<str> = Arc::from("orders.ledger");
        let audit: Arc<str> = Arc::from("orders.audit");
        let queued = Instant::now();
        let first_at = |index: usize| match index {
            0 => Some((&ledger, queued)),
            2 => Some((&audit, queued)),
            _ => None,
        };
        let reached = |station: &Arc<str>| Reached {
            station: Arc::clone(station),
            queued_at: queued,
        };
        assert_eq!(
            gather(3, 2, first_at)?,
            Some(Arc::from(vec![reached(&ledger), reached(&audit)]))
        );
        assert_eq!(gather(3, 0, |_| None)?, None);
        Ok(())
    }
}