liminal_server/tap/subscription/

publish.rs

//! The existing channel publish hook and its measured fan-out.

use super::{
    Admitted, Arc, Hop, Instant, Ordering, PublishFact, Reached, Recipient, Tap, TapError,
    TapIdent, TapMessage, TapSource, elapsed_ms, gather, lock,
};

impl Tap {
    /// 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.
    pub(super) 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,
                call: None,
                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)),
        }
    }
}