dwara_core/lifecycle/
journey.rs

1//! API journey recorder (DW-110): records the request flow through the
2//! gateway as a JSON document for debugging.
3//!
4//! A [`Journey`] is the ordered sequence of phases a request passed
5//! through (route match, authn, authz, transforms, upstream pick,
6//! response), each captured as a [`JourneyStep`] (phase, duration,
7//! result, detail). The recorder is a bounded in-memory ring buffer of
8//! recent journeys (for live debugging) plus a JSON serializer that
9//! stores the journey via the existing analytics raw table's custom
10//! dimensions column (a JSON object column -- no schema change needed).
11//!
12//! ## Storage
13//!
14//! The journey JSON document is carried on the
15//! `AccessRecord`(crate::observability::AccessRecord)'s `custom`
16//! dimensions as a `_journey` key (the same JSON object column the
17//! DW-043 custom dimensions ride). This reuses the existing raw
18//! table's fire-and-forget write path -- no schema migration, no new
19//! table, no new writer channel. The in-memory ring buffer is a
20//! bounded `VecDeque` (capped at `JOURNEY_BUFFER_CAP`) for live
21//! debugging queries; the durable copy is the raw row.
22//!
23//! ## Retention
24//!
25//! The journey's durable retention is the raw table's retention (the
26//! `analytics.retention` config, default 24h). The `retention_hours`
27//! config field on [`JourneyConfig`] is an advisory cap that the
28//! recorder honors by dropping journeys older than the window from the
29//! in-memory buffer (the raw table's own retention handles the durable
30//! copy).
31//!
32//! The config schema type ([`JourneyConfig`]) lives in
33//! [`crate::config`] (always present, so configs round-trip without
34//! the `api_lifecycle` feature). This module re-exports it as the
35//! runtime-facing alias the recorder consumes.
36
37use std::collections::VecDeque;
38use std::sync::Mutex;
39
40use serde::{Deserialize, Serialize};
41use serde_json::Value;
42
43pub use crate::config::LifecycleJourneyConfig as JourneyConfig;
44
45/// The result of a journey step.
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
47#[serde(rename_all = "lowercase")]
48pub enum JourneyStepResult {
49    Success,
50    Failure,
51}
52
53/// One step in a request's journey through the gateway.
54#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
55#[serde(deny_unknown_fields)]
56pub struct JourneyStep {
57    /// The phase name (e.g. `route_match`, `authn`, `authz`,
58    /// `transforms`, `upstream_pick`, `response`).
59    pub phase: String,
60    /// The duration of this phase in milliseconds.
61    pub duration_ms: f64,
62    /// The outcome of this phase.
63    pub result: JourneyStepResult,
64    /// A free-form detail string (e.g. the matched route name, the
65    /// picked upstream, the response status). Never secrets -- the
66    /// same redaction posture as the access log applies.
67    #[serde(skip_serializing_if = "Option::is_none")]
68    pub detail: Option<String>,
69}
70
71/// A recorded request journey: the request id, the ordered steps, and
72/// the timestamp (wall-clock ms since the Unix epoch).
73#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
74#[serde(deny_unknown_fields)]
75pub struct Journey {
76    pub request_id: String,
77    pub steps: Vec<JourneyStep>,
78    pub timestamp_ms: u64,
79}
80
81impl Journey {
82    /// Create a new empty journey for the given request id.
83    pub fn new(request_id: impl Into<String>) -> Self {
84        Self {
85            request_id: request_id.into(),
86            steps: Vec::new(),
87            timestamp_ms: now_ms(),
88        }
89    }
90
91    /// Record a step in the journey.
92    pub fn add_step(&mut self, step: JourneyStep) {
93        self.steps.push(step);
94    }
95
96    /// Record a step with the given phase, result, and detail. The
97    /// duration is measured from the `started` instant to now.
98    pub fn add_timed_step(
99        &mut self,
100        phase: impl Into<String>,
101        started: std::time::Instant,
102        result: JourneyStepResult,
103        detail: Option<String>,
104    ) {
105        let duration_ms = started.elapsed().as_secs_f64() * 1000.0;
106        self.add_step(JourneyStep {
107            phase: phase.into(),
108            duration_ms,
109            result,
110            detail,
111        });
112    }
113
114    /// Serialize the journey to a JSON string.
115    pub fn to_json(&self) -> Result<String, String> {
116        serde_json::to_string(self).map_err(|e| format!("journey serialize failed: {e}"))
117    }
118
119    /// Serialize the journey to a [`serde_json::Value`].
120    pub fn to_value(&self) -> Result<Value, String> {
121        serde_json::to_value(self).map_err(|e| format!("journey serialize failed: {e}"))
122    }
123
124    /// The total duration of all steps (sum of step durations).
125    pub fn total_duration_ms(&self) -> f64 {
126        self.steps.iter().map(|s| s.duration_ms).sum()
127    }
128
129    /// Whether any step failed.
130    pub fn has_failure(&self) -> bool {
131        self.steps
132            .iter()
133            .any(|s| s.result == JourneyStepResult::Failure)
134    }
135}
136
137/// The in-memory ring buffer cap (recent journeys for live debugging).
138const JOURNEY_BUFFER_CAP: usize = 4096;
139
140/// The journey recorder: a bounded in-memory ring buffer of recent
141/// journeys, plus the retention window for dropping stale entries.
142#[derive(Debug)]
143pub struct JourneyRecorder {
144    config: JourneyConfig,
145    buffer: Mutex<VecDeque<Journey>>,
146}
147
148impl JourneyRecorder {
149    /// Build a recorder from the config.
150    pub fn new(config: JourneyConfig) -> Self {
151        Self {
152            config,
153            buffer: Mutex::new(VecDeque::new()),
154        }
155    }
156
157    /// Whether recording is enabled.
158    pub fn enabled(&self) -> bool {
159        self.config.enabled
160    }
161
162    /// The configured retention window in hours.
163    pub fn retention_hours(&self) -> u64 {
164        self.config.retention_hours
165    }
166
167    /// Record a completed journey into the in-memory ring buffer. The
168    /// caller is responsible for also stamping the journey onto the
169    /// `AccessRecord`'s custom dimensions (via [`journey_dimension`])
170    /// so the durable copy rides the raw table's write path. This
171    /// method never blocks: the buffer is a short mutex over a capped
172    /// `VecDeque`; when full, the oldest entry is evicted.
173    pub fn record(&self, journey: Journey) {
174        if !self.config.enabled {
175            return;
176        }
177        let mut buf = self.buffer.lock().expect("journey buffer mutex poisoned");
178        // Drop stale entries (older than the retention window).
179        let cutoff = now_ms().saturating_sub(self.config.retention_hours * 3600 * 1000);
180        while let Some(front) = buf.front() {
181            if front.timestamp_ms < cutoff {
182                buf.pop_front();
183            } else {
184                break;
185            }
186        }
187        // Evict the oldest when at cap.
188        if buf.len() >= JOURNEY_BUFFER_CAP {
189            buf.pop_front();
190        }
191        buf.push_back(journey);
192    }
193
194    /// Snapshot the in-memory journeys (newest first). Never blocks
195    /// beyond the short mutex hold.
196    pub fn snapshot(&self) -> Vec<Journey> {
197        let buf = self.buffer.lock().expect("journey buffer mutex poisoned");
198        let mut out: Vec<Journey> = buf.iter().rev().cloned().collect();
199        // Drop stale entries on read too (a long-idle buffer can hold
200        // stale entries when no new recording evicted them).
201        let cutoff = now_ms().saturating_sub(self.config.retention_hours * 3600 * 1000);
202        out.retain(|j| j.timestamp_ms >= cutoff);
203        out
204    }
205
206    /// The number of journeys currently in the buffer.
207    pub fn len(&self) -> usize {
208        self.buffer
209            .lock()
210            .expect("journey buffer mutex poisoned")
211            .len()
212    }
213
214    /// Whether the buffer is empty.
215    pub fn is_empty(&self) -> bool {
216        self.len() == 0
217    }
218}
219
220/// Build the custom-dimension entry that carries a journey on an
221/// `AccessRecord` into the raw table's JSON object column. The key
222/// is `_journey` (the leading underscore avoids collisions with
223/// operator-declared dimensions). Returns the `(key, value)` pair to
224/// push onto `AccessRecord::custom`.
225pub fn journey_dimension(journey: &Journey) -> Result<(String, String), String> {
226    let json = journey.to_json()?;
227    Ok(("_journey".to_string(), json))
228}
229
230/// Wall-clock milliseconds since the Unix epoch.
231fn now_ms() -> u64 {
232    std::time::SystemTime::now()
233        .duration_since(std::time::UNIX_EPOCH)
234        .map(|d| d.as_millis() as u64)
235        .unwrap_or(0)
236}
dwara_core/lifecycle/journey.rs

dwara_core/lifecycle/
journey.rs
