pb_mapper_client/
diagnostics.rs

1//! Bounded, credential-free recovery counters exposed separately from lifecycle status.
2//!
3//! Besides counters, a tunnel keeps the direct attempt it is making right now
4//! and a short history of what happened to it ([`TunnelEvent`]), so a caller can
5//! show recovery as it unfolds instead of a bare "retrying". Both are written
6//! only on control-path transitions, never per packet.
7
8use crate::endpoint::RelayEndpoint;
9use serde::Serialize;
10use std::collections::VecDeque;
11use std::sync::{Arc, Mutex};
12use tokio::time::{Duration, Instant};
13
14/// The phase most recently entered by a tunnel worker.
15#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
16#[serde(rename_all = "snake_case")]
17pub enum RecoveryPhase {
18    Starting,
19    Capacity,
20    Resolving,
21    Connecting,
22    Handshake,
23    Ready,
24    Backoff,
25    Stopped,
26}
27
28impl RecoveryPhase {
29    /// Stable diagnostic label, matching the serialized representation.
30    pub fn as_str(self) -> &'static str {
31        match self {
32            Self::Starting => "starting",
33            Self::Capacity => "capacity",
34            Self::Resolving => "resolving",
35            Self::Connecting => "connecting",
36            Self::Handshake => "handshake",
37            Self::Ready => "ready",
38            Self::Backoff => "backoff",
39            Self::Stopped => "stopped",
40        }
41    }
42}
43
44/// Why a recovery attempt failed, without credentials or application data.
45#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
46#[serde(rename_all = "snake_case")]
47pub enum RecoveryFailure {
48    Dns,
49    Transport,
50    Timeout,
51    ServiceUnavailable,
52    Rejected,
53    NetworkChanged,
54}
55
56impl RecoveryFailure {
57    /// Stable diagnostic label, matching the serialized representation.
58    pub fn as_str(self) -> &'static str {
59        match self {
60            Self::Dns => "dns",
61            Self::Transport => "transport",
62            Self::Timeout => "timeout",
63            Self::ServiceUnavailable => "service_unavailable",
64            Self::Rejected => "rejected",
65            Self::NetworkChanged => "network_changed",
66        }
67    }
68}
69
70/// Where a direct attempt stands. Every stage before [`Self::Committing`] can
71/// still fall back to the relay; a fallback names the stage it left.
72#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
73#[serde(rename_all = "snake_case")]
74pub enum DirectStage {
75    /// Asking the relay whether the publisher offers direct, and opening the
76    /// control link.
77    Negotiating,
78    /// Learning this endpoint's public address and requesting a session grant.
79    Discovering,
80    /// Dialing the publisher through both NATs.
81    Punching,
82    /// The path is up; the publisher is opening its local service.
83    Committing,
84}
85
86impl DirectStage {
87    /// Stable diagnostic label, matching the serialized representation.
88    pub fn as_str(self) -> &'static str {
89        match self {
90            Self::Negotiating => "negotiating",
91            Self::Discovering => "discovering",
92            Self::Punching => "punching",
93            Self::Committing => "committing",
94        }
95    }
96}
97
98/// The direct attempt a tunnel is making right now.
99#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
100pub struct DirectAttempt {
101    /// The stage most recently entered by any attempt in flight.
102    pub stage: DirectStage,
103    /// Time since that attempt began.
104    pub elapsed_ms: u64,
105    /// Attempts in flight, this one included.
106    pub in_flight: u32,
107}
108
109/// What happened, in a [`TunnelEvent`].
110#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
111#[serde(tag = "kind", rename_all = "snake_case")]
112pub enum TunnelEventKind {
113    /// The relay answered for the first time, or again after failures.
114    RelayReady { latency_ms: u64 },
115    /// A relay attempt failed; the next one is `retry_in_ms` away (zero when
116    /// the worker retries at once or the failure is final).
117    RelayFailed {
118        failure: RecoveryFailure,
119        retry_in_ms: u64,
120    },
121    /// The local network changed; recovery restarts at once.
122    NetworkChanged,
123    /// A session took the direct path.
124    DirectEstablished { setup_ms: u64 },
125    /// A session gave up on direct at `stage` and used the relay.
126    DirectFallback { stage: DirectStage },
127}
128
129impl TunnelEventKind {
130    /// Whether a repeat of `self` folds into the entry for `other`.
131    fn folds_into(&self, other: &Self) -> bool {
132        match (self, other) {
133            (Self::RelayFailed { failure: a, .. }, Self::RelayFailed { failure: b, .. }) => a == b,
134            (Self::DirectFallback { stage: a }, Self::DirectFallback { stage: b }) => a == b,
135            (Self::RelayReady { .. }, Self::RelayReady { .. })
136            | (Self::NetworkChanged, Self::NetworkChanged)
137            | (Self::DirectEstablished { .. }, Self::DirectEstablished { .. }) => true,
138            _ => false,
139        }
140    }
141}
142
143/// One entry of a tunnel's recent history, oldest first in
144/// [`TunnelDiagnostics::events`]. Consecutive repeats fold into one entry, so
145/// a long outage cannot push out what came before it.
146#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)]
147pub struct TunnelEvent {
148    /// Increases with every recorded event, a fold included, so a poller can
149    /// tell what it has not seen yet.
150    pub seq: u64,
151    /// Time since the latest occurrence.
152    pub age_ms: u64,
153    /// Occurrences this entry stands for.
154    pub repeats: u32,
155    /// The latest occurrence.
156    pub kind: TunnelEventKind,
157}
158
159/// How many entries a tunnel's history keeps.
160const EVENT_CAPACITY: usize = 16;
161
162/// A point-in-time view of this tunnel and its process-shared relay budget.
163/// Non-exhaustive: fields are added as the SDK observes more.
164#[derive(Clone, Debug, Serialize)]
165#[non_exhaustive]
166pub struct TunnelDiagnostics {
167    /// The carrier this tunnel reaches the relay over, fixed when it started.
168    pub relay_transport: crate::relay::RelayTransport,
169    #[serde(skip_serializing_if = "Option::is_none")]
170    pub direct: Option<PathDiagnostics>,
171    pub sdk_version: &'static str,
172    /// Most recent worker phase; use tunnel status for aggregate readiness.
173    pub last_attempt_phase: RecoveryPhase,
174    pub attempts: u64,
175    pub consecutive_failures: u64,
176    pub last_failure: Option<RecoveryFailure>,
177    /// Time since the last authenticated relay reply, even a negative reply.
178    pub last_success_age_ms: Option<u64>,
179    /// Last answered setup latency, excluding admission queue time.
180    pub last_setup_latency_ms: Option<u64>,
181    pub next_retry_in_ms: Option<u64>,
182    pub dns_age_ms: Option<u64>,
183    pub network_generation: u64,
184    /// Shared across this process for the same configured relay address.
185    pub active_control_setups: usize,
186    pub active_data_setups: usize,
187    /// Recent history, oldest first; at most 16 entries.
188    pub events: Vec<TunnelEvent>,
189}
190
191#[derive(Clone, Debug, Default, Serialize)]
192pub struct PathDiagnostics {
193    pub direct_sessions: u64,
194    pub relay_sessions: u64,
195    pub active_direct: u64,
196    pub active_relay: u64,
197    pub fallbacks: u64,
198    pub last_fallback: Option<String>,
199    /// The direct attempt in flight, if any.
200    #[serde(skip_serializing_if = "Option::is_none")]
201    pub attempt: Option<DirectAttempt>,
202}
203
204#[derive(Clone, Copy)]
205struct Event {
206    seq: u64,
207    at: Instant,
208    repeats: u32,
209    kind: TunnelEventKind,
210}
211
212struct State {
213    direct: Option<PathDiagnostics>,
214    phase: RecoveryPhase,
215    attempts: u64,
216    failures: u64,
217    failure: Option<RecoveryFailure>,
218    success: Option<Instant>,
219    latency: Option<Duration>,
220    retry: Option<Instant>,
221    /// Whether the last relay attempt succeeded, so only a change is recorded.
222    ready: bool,
223    /// The network generation last seen, so observers of one change record
224    /// it once and a worker starting on an old one records nothing.
225    network: Option<u64>,
226    /// The stage and start of the latest direct attempt still in flight.
227    attempt: Option<(DirectStage, Instant)>,
228    in_flight: u32,
229    seq: u64,
230    events: VecDeque<Event>,
231}
232
233impl State {
234    fn record(&mut self, kind: TunnelEventKind) {
235        self.seq = self.seq.wrapping_add(1);
236        let event = Event {
237            seq: self.seq,
238            at: Instant::now(),
239            repeats: 1,
240            kind,
241        };
242        if let Some(last) = self
243            .events
244            .back_mut()
245            .filter(|last| kind.folds_into(&last.kind))
246        {
247            *last = Event {
248                repeats: last.repeats.saturating_add(1),
249                ..event
250            };
251            return;
252        }
253        if self.events.len() == EVENT_CAPACITY {
254            self.events.pop_front();
255        }
256        self.events.push_back(event);
257    }
258}
259
260#[derive(Clone)]
261pub(crate) struct Diagnostics(Arc<Mutex<State>>);
262
263impl Default for Diagnostics {
264    fn default() -> Self {
265        Self(Arc::new(Mutex::new(State {
266            direct: None,
267            phase: RecoveryPhase::Starting,
268            attempts: 0,
269            failures: 0,
270            failure: None,
271            success: None,
272            latency: None,
273            retry: None,
274            ready: false,
275            network: None,
276            attempt: None,
277            in_flight: 0,
278            seq: 0,
279            events: VecDeque::new(),
280        })))
281    }
282}
283
284impl Diagnostics {
285    fn state(&self) -> std::sync::MutexGuard<'_, State> {
286        self.0.lock().unwrap_or_else(|e| e.into_inner())
287    }
288    pub(crate) fn has_active_direct(&self) -> bool {
289        self.state()
290            .direct
291            .as_ref()
292            .is_some_and(|p| p.active_direct != 0)
293    }
294    #[cfg(feature = "p2p")]
295    pub(crate) fn path_started(&self, path: pb_mapper_protocol::direct::SessionPath) {
296        let mut state = self.state();
297        let p = state.direct.get_or_insert_default();
298        match path {
299            pb_mapper_protocol::direct::SessionPath::Direct => {
300                p.direct_sessions += 1;
301                p.active_direct += 1;
302            }
303            pb_mapper_protocol::direct::SessionPath::Relay => {
304                p.relay_sessions += 1;
305                p.active_relay += 1;
306            }
307        }
308    }
309    #[cfg(feature = "p2p")]
310    pub(crate) fn path_closed(&self, path: pb_mapper_protocol::direct::SessionPath) {
311        let mut state = self.state();
312        let p = state.direct.get_or_insert_default();
313        match path {
314            pb_mapper_protocol::direct::SessionPath::Direct => {
315                p.active_direct = p.active_direct.saturating_sub(1)
316            }
317            pb_mapper_protocol::direct::SessionPath::Relay => {
318                p.active_relay = p.active_relay.saturating_sub(1)
319            }
320        }
321    }
322    /// Begin a direct attempt; the guard reports its stages and outcome.
323    #[cfg(feature = "p2p")]
324    pub(crate) fn direct_attempt(&self) -> DirectAttemptGuard {
325        let started = Instant::now();
326        let mut state = self.state();
327        // The attempt rides on the path view, so it shows from the first one.
328        state.direct.get_or_insert_default();
329        state.in_flight = state.in_flight.saturating_add(1);
330        state.attempt = Some((DirectStage::Negotiating, started));
331        DirectAttemptGuard {
332            diagnostics: self.clone(),
333            stage: DirectStage::Negotiating,
334            started,
335        }
336    }
337
338    pub(crate) fn heard(&self) {
339        self.state().success = Some(Instant::now());
340    }
341    pub(crate) fn responded(&self, elapsed: Duration) {
342        let mut state = self.state();
343        state.success = Some(Instant::now());
344        state.latency = Some(elapsed);
345    }
346    pub(crate) fn phase(&self, phase: RecoveryPhase) {
347        self.state().phase = phase;
348    }
349    pub(crate) fn attempt(&self) {
350        let mut state = self.state();
351        state.attempts = state.attempts.saturating_add(1);
352        state.retry = None;
353        state.phase = RecoveryPhase::Capacity;
354    }
355    pub(crate) fn succeeded(&self, elapsed: Duration) {
356        let mut state = self.state();
357        state.phase = RecoveryPhase::Ready;
358        state.failures = 0;
359        state.success = Some(Instant::now());
360        state.latency = Some(elapsed);
361        state.retry = None;
362        if !state.ready {
363            state.ready = true;
364            state.record(TunnelEventKind::RelayReady {
365                latency_ms: ms(elapsed),
366            });
367        }
368    }
369    pub(crate) fn failed(&self, failure: RecoveryFailure, retry: Duration) -> bool {
370        let mut state = self.state();
371        state.failures = state.failures.saturating_add(1);
372        let emit = state.failure != Some(failure) || state.failures.is_power_of_two();
373        state.phase = RecoveryPhase::Backoff;
374        state.failure = Some(failure);
375        state.retry = Some(Instant::now() + retry);
376        state.ready = false;
377        // A network change is recorded once, by whoever observes it first.
378        if failure != RecoveryFailure::NetworkChanged {
379            state.record(TunnelEventKind::RelayFailed {
380                failure,
381                retry_in_ms: ms(retry),
382            });
383        }
384        emit
385    }
386    /// Note the network generation a worker starts on.
387    pub(crate) fn network_seen(&self, generation: u64) {
388        self.state().network.get_or_insert(generation);
389    }
390    /// Record a change of the local network to `generation`, once however
391    /// many parts of this tunnel observe it.
392    pub(crate) fn network_changed(&self, generation: u64) {
393        let mut state = self.state();
394        if state.network != Some(generation) {
395            state.network = Some(generation);
396            state.record(TunnelEventKind::NetworkChanged);
397        }
398    }
399    pub(crate) fn snapshot(&self, endpoint: &RelayEndpoint) -> TunnelDiagnostics {
400        let state = self.state();
401        let now = Instant::now();
402        let (active_control_setups, active_data_setups) = endpoint.active_setups();
403        let attempt = state.attempt.map(|(stage, started)| DirectAttempt {
404            stage,
405            elapsed_ms: ms(now.saturating_duration_since(started)),
406            in_flight: state.in_flight,
407        });
408        TunnelDiagnostics {
409            relay_transport: endpoint.transport(),
410            direct: state
411                .direct
412                .clone()
413                .map(|direct| PathDiagnostics { attempt, ..direct }),
414            sdk_version: env!("CARGO_PKG_VERSION"),
415            last_attempt_phase: state.phase,
416            attempts: state.attempts,
417            consecutive_failures: state.failures,
418            last_failure: state.failure,
419            last_success_age_ms: state
420                .success
421                .map(|time| ms(now.saturating_duration_since(time))),
422            last_setup_latency_ms: state.latency.map(ms),
423            next_retry_in_ms: state
424                .retry
425                .map(|time| ms(time.saturating_duration_since(now))),
426            dns_age_ms: endpoint.dns_age().map(ms),
427            network_generation: endpoint.network_generation(),
428            active_control_setups,
429            active_data_setups,
430            events: state
431                .events
432                .iter()
433                .map(|event| TunnelEvent {
434                    seq: event.seq,
435                    age_ms: ms(now.saturating_duration_since(event.at)),
436                    repeats: event.repeats,
437                    kind: event.kind,
438                })
439                .collect(),
440        }
441    }
442}
443
444fn ms(time: Duration) -> u64 {
445    time.as_millis().min(u128::from(u64::MAX)) as u64
446}
447
448/// One direct attempt in flight. Dropping it without an outcome (an error
449/// after commit, a cancelled setup) records nothing.
450#[cfg(feature = "p2p")]
451pub(crate) struct DirectAttemptGuard {
452    diagnostics: Diagnostics,
453    stage: DirectStage,
454    started: Instant,
455}
456
457#[cfg(feature = "p2p")]
458impl DirectAttemptGuard {
459    pub(crate) fn stage(&mut self, stage: DirectStage) {
460        self.stage = stage;
461        self.diagnostics.state().attempt = Some((stage, self.started));
462    }
463    pub(crate) fn established(self) {
464        let setup_ms = ms(self.started.elapsed());
465        self.diagnostics
466            .state()
467            .record(TunnelEventKind::DirectEstablished { setup_ms });
468    }
469    /// Fall back to the relay for `reason`, the legacy label kept in
470    /// [`PathDiagnostics::last_fallback`].
471    pub(crate) fn fell_back(self, reason: &str) {
472        let mut state = self.diagnostics.state();
473        let p = state.direct.get_or_insert_default();
474        p.fallbacks += 1;
475        p.last_fallback = Some(reason.into());
476        state.record(TunnelEventKind::DirectFallback { stage: self.stage });
477    }
478}
479
480#[cfg(feature = "p2p")]
481impl Drop for DirectAttemptGuard {
482    fn drop(&mut self) {
483        let mut state = self.diagnostics.state();
484        state.in_flight = state.in_flight.saturating_sub(1);
485        if state.in_flight == 0 {
486            state.attempt = None;
487        }
488    }
489}
490
491#[cfg(feature = "p2p")]
492pub(crate) struct PathUsage(pub Diagnostics, pub pb_mapper_protocol::direct::SessionPath);
493#[cfg(feature = "p2p")]
494impl Drop for PathUsage {
495    fn drop(&mut self) {
496        self.0.path_closed(self.1);
497    }
498}
499
500#[cfg(test)]
501#[allow(clippy::unwrap_used)]
502mod tests {
503    use super::*;
504
505    fn kinds(diagnostics: &Diagnostics) -> Vec<(TunnelEventKind, u32)> {
506        diagnostics
507            .snapshot(&RelayEndpoint::shared("127.0.0.1:1"))
508            .events
509            .into_iter()
510            .map(|event| (event.kind, event.repeats))
511            .collect()
512    }
513
514    #[tokio::test]
515    async fn history_records_changes_and_folds_repeats() {
516        let diagnostics = Diagnostics::default();
517        diagnostics.succeeded(Duration::from_millis(40));
518        // Steady success is not news: only a change is recorded.
519        diagnostics.succeeded(Duration::from_millis(41));
520        diagnostics.failed(RecoveryFailure::Transport, Duration::from_millis(200));
521        diagnostics.failed(RecoveryFailure::Transport, Duration::from_millis(400));
522        diagnostics.network_changed(1);
523        diagnostics.network_changed(1);
524        diagnostics.failed(RecoveryFailure::NetworkChanged, Duration::ZERO);
525        diagnostics.succeeded(Duration::from_millis(55));
526        assert_eq!(
527            kinds(&diagnostics),
528            vec![
529                (TunnelEventKind::RelayReady { latency_ms: 40 }, 1),
530                (
531                    TunnelEventKind::RelayFailed {
532                        failure: RecoveryFailure::Transport,
533                        retry_in_ms: 400
534                    },
535                    2
536                ),
537                (TunnelEventKind::NetworkChanged, 1),
538                (TunnelEventKind::RelayReady { latency_ms: 55 }, 1),
539            ]
540        );
541    }
542
543    #[tokio::test]
544    async fn history_is_bounded_and_sequenced() {
545        let diagnostics = Diagnostics::default();
546        for generation in 1..=40 {
547            diagnostics.network_changed(generation);
548            diagnostics.failed(RecoveryFailure::Timeout, Duration::ZERO);
549        }
550        let events = diagnostics
551            .snapshot(&RelayEndpoint::shared("127.0.0.1:1"))
552            .events;
553        assert_eq!(events.len(), EVENT_CAPACITY);
554        assert!(events.windows(2).all(|pair| pair[0].seq < pair[1].seq));
555    }
556
557    #[cfg(feature = "p2p")]
558    #[tokio::test]
559    async fn a_direct_attempt_shows_its_stage_until_it_settles() {
560        let diagnostics = Diagnostics::default();
561        let endpoint = RelayEndpoint::shared("127.0.0.1:1");
562        let attempt_of = |d: &Diagnostics| d.snapshot(&endpoint).direct.and_then(|p| p.attempt);
563        let mut first = diagnostics.direct_attempt();
564        first.stage(DirectStage::Punching);
565        assert_eq!(
566            attempt_of(&diagnostics).unwrap().stage,
567            DirectStage::Punching
568        );
569        let second = diagnostics.direct_attempt();
570        let seen = attempt_of(&diagnostics).unwrap();
571        assert_eq!((seen.stage, seen.in_flight), (DirectStage::Negotiating, 2));
572        first.fell_back("direct_timeout_or_unreachable");
573        assert_eq!(attempt_of(&diagnostics).unwrap().in_flight, 1);
574        second.established();
575        assert_eq!(attempt_of(&diagnostics), None);
576        let direct = diagnostics.snapshot(&endpoint).direct.unwrap();
577        assert_eq!(direct.fallbacks, 1);
578        assert_eq!(
579            direct.last_fallback.as_deref(),
580            Some("direct_timeout_or_unreachable")
581        );
582        assert!(matches!(
583            kinds(&diagnostics)[..],
584            [
585                (
586                    TunnelEventKind::DirectFallback {
587                        stage: DirectStage::Punching
588                    },
589                    1
590                ),
591                (TunnelEventKind::DirectEstablished { .. }, 1),
592            ]
593        ));
594    }
595}
pb_mapper_client/diagnostics.rs

pb_mapper_client/
diagnostics.rs
