liminal-server 0.24.9

//! A completed call is sampled once, with its measured request and reply journey.

use super::{Admitted, Hop, Tap, TapError, TapIdent, TapMessage, TapSource, lock};
use crate::tap::message::{CallDetails, CallFact, Reached};
use std::sync::{Arc, atomic::Ordering};
use std::time::Instant;

impl Tap {
    /// The station already registered for this source.
    ///
    /// # Errors
    /// The station registry is poisoned.
    pub fn source_station(&self, source: TapSource) -> Result<Option<Arc<str>>, TapError> {
        self.station_of(source)
    }

    /// Offers one call's terminal outcome to the same sampled rings as publishes.
    /// No payload is retained, and the drainer is told after every lock is released.
    ///
    /// # Errors
    /// A tap lock is poisoned, or a measured hop cannot be represented.
    pub fn observe_call(&self, fact: &CallFact<'_>) -> Result<bool, TapError> {
        if self.shared.active.load(Ordering::Acquire) == 0 {
            return Ok(false);
        }
        let observation = self.shared.observed.fetch_add(1, Ordering::Relaxed);
        let mut kept = false;
        let result = self.keep_call(fact, observation, &mut kept);
        if kept && let Some(waker) = self.shared.waker.get() {
            waker();
        }
        result.map(|()| kept)
    }

    fn keep_call(
        &self,
        fact: &CallFact<'_>,
        observation: u64,
        kept: &mut bool,
    ) -> Result<(), TapError> {
        let terminal_ms = call_ms(fact.received_at, fact.finished_at, 0)?;
        let served_ms = call_ms(fact.received_at, fact.forwarded_at, 1)?;
        if served_ms > terminal_ms {
            return Err(TapError::HopsOutOfOrder {
                hop: 1,
                at_ms: served_ms,
                before_ms: terminal_ms,
            });
        }
        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 mut hops = Vec::new();
            if let Some(station) = fact.station {
                hops.push(Hop {
                    node: Arc::clone(station),
                    at_ms: 0,
                });
            }
            hops.push(Hop {
                node: Arc::clone(&self.node),
                at_ms: 0,
            });
            let reached: Arc<[Reached]> = fact.served_by.map_or_else(
                || Arc::clone(&self.shared.no_stations),
                |station| {
                    hops.push(Hop {
                        node: Arc::clone(station),
                        at_ms: served_ms,
                    });
                    Arc::from([Reached {
                        station: Arc::clone(station),
                        queued_at: fact.forwarded_at,
                    }])
                },
            );
            hops.push(Hop {
                node: Arc::clone(&self.node),
                at_ms: terminal_ms,
            });
            let message = TapMessage {
                ident: fact
                    .operation
                    .map_or(TapIdent::Observation(observation), |operation| {
                        TapIdent::Call {
                            connection: fact.connection,
                            operation,
                        }
                    }),
                subject: fact.subject.to_owned(),
                size: fact.size,
                outcome: fact.outcome,
                seen_at: fact.received_at,
                source: TapSource::Connection(fact.connection),
                station: fact.station.cloned(),
                to_stations: Some(reached),
                hops,
                reply_to: None,
                held: false,
                payload: None,
                call: Some(CallDetails {
                    latency_ms: (fact.outcome == crate::tap::TapOutcome::Answered)
                        .then_some(terminal_ms),
                    refusal: fact.refusal.map(str::to_owned),
                }),
            };
            let one_in = state.one_in;
            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);
        Ok(())
    }
}

fn call_ms(start: Instant, end: Instant, hop: usize) -> Result<u64, TapError> {
    let elapsed =
        end.checked_duration_since(start)
            .ok_or_else(|| TapError::HopBeforeFirstSight {
                hop,
                early_ms: u64::try_from(start.duration_since(end).as_millis()).unwrap_or(u64::MAX),
            })?;
    u64::try_from(elapsed.as_millis()).map_err(|_| TapError::HopOverflow { hop })
}