liminal_server/tap/subscription/
call.rs

1//! A completed call is sampled once, with its measured request and reply journey.
2
3use super::{Admitted, Hop, Tap, TapError, TapIdent, TapMessage, TapSource, lock};
4use crate::tap::message::{CallDetails, CallFact, Reached};
5use std::sync::{Arc, atomic::Ordering};
6use std::time::Instant;
7
8impl Tap {
9    /// The station already registered for this source.
10    ///
11    /// # Errors
12    /// The station registry is poisoned.
13    pub fn source_station(&self, source: TapSource) -> Result<Option<Arc<str>>, TapError> {
14        self.station_of(source)
15    }
16
17    /// Offers one call's terminal outcome to the same sampled rings as publishes.
18    /// No payload is retained, and the drainer is told after every lock is released.
19    ///
20    /// # Errors
21    /// A tap lock is poisoned, or a measured hop cannot be represented.
22    pub fn observe_call(&self, fact: &CallFact<'_>) -> Result<bool, TapError> {
23        if self.shared.active.load(Ordering::Acquire) == 0 {
24            return Ok(false);
25        }
26        let observation = self.shared.observed.fetch_add(1, Ordering::Relaxed);
27        let mut kept = false;
28        let result = self.keep_call(fact, observation, &mut kept);
29        if kept && let Some(waker) = self.shared.waker.get() {
30            waker();
31        }
32        result.map(|()| kept)
33    }
34
35    fn keep_call(
36        &self,
37        fact: &CallFact<'_>,
38        observation: u64,
39        kept: &mut bool,
40    ) -> Result<(), TapError> {
41        let terminal_ms = call_ms(fact.received_at, fact.finished_at, 0)?;
42        let served_ms = call_ms(fact.received_at, fact.forwarded_at, 1)?;
43        if served_ms > terminal_ms {
44            return Err(TapError::HopsOutOfOrder {
45                hop: 1,
46                at_ms: served_ms,
47                before_ms: terminal_ms,
48            });
49        }
50        let subscribers = lock(&self.shared.subscribers, "tap subscribers")?;
51        for subscriber in subscribers.iter() {
52            if !subscriber.filter.matches(fact.subject) {
53                continue;
54            }
55            let mut state = lock(&subscriber.state, "tap subscriber")?;
56            if !state.admits() {
57                continue;
58            }
59            let mut hops = Vec::new();
60            if let Some(station) = fact.station {
61                hops.push(Hop {
62                    node: Arc::clone(station),
63                    at_ms: 0,
64                });
65            }
66            hops.push(Hop {
67                node: Arc::clone(&self.node),
68                at_ms: 0,
69            });
70            let reached: Arc<[Reached]> = fact.served_by.map_or_else(
71                || Arc::clone(&self.shared.no_stations),
72                |station| {
73                    hops.push(Hop {
74                        node: Arc::clone(station),
75                        at_ms: served_ms,
76                    });
77                    Arc::from([Reached {
78                        station: Arc::clone(station),
79                        queued_at: fact.forwarded_at,
80                    }])
81                },
82            );
83            hops.push(Hop {
84                node: Arc::clone(&self.node),
85                at_ms: terminal_ms,
86            });
87            let message = TapMessage {
88                ident: fact
89                    .operation
90                    .map_or(TapIdent::Observation(observation), |operation| {
91                        TapIdent::Call {
92                            connection: fact.connection,
93                            operation,
94                        }
95                    }),
96                subject: fact.subject.to_owned(),
97                size: fact.size,
98                outcome: fact.outcome,
99                seen_at: fact.received_at,
100                source: TapSource::Connection(fact.connection),
101                station: fact.station.cloned(),
102                to_stations: Some(reached),
103                hops,
104                reply_to: None,
105                held: false,
106                payload: None,
107                call: Some(CallDetails {
108                    latency_ms: (fact.outcome == crate::tap::TapOutcome::Answered)
109                        .then_some(terminal_ms),
110                    refusal: fact.refusal.map(str::to_owned),
111                }),
112            };
113            let one_in = state.one_in;
114            if state.ring.push(Admitted { one_in, message }) {
115                self.shared.dropped_total.fetch_add(1, Ordering::Relaxed);
116            }
117            state.after_push();
118            drop(state);
119            *kept = true;
120        }
121        drop(subscribers);
122        Ok(())
123    }
124}
125
126fn call_ms(start: Instant, end: Instant, hop: usize) -> Result<u64, TapError> {
127    let elapsed =
128        end.checked_duration_since(start)
129            .ok_or_else(|| TapError::HopBeforeFirstSight {
130                hop,
131                early_ms: u64::try_from(start.duration_since(end).as_millis()).unwrap_or(u64::MAX),
132            })?;
133    u64::try_from(elapsed.as_millis()).map_err(|_| TapError::HopOverflow { hop })
134}
liminal_server/tap/subscription/call.rs

liminal_server/tap/subscription/
call.rs
