apimock_server/
trace.rs

1//! Live match-trace channel — RFC 006 (in-process) + RFC 009 (transport).
2//!
3//! # Architecture
4//!
5//! ```text
6//!  HTTP handler ──► TraceEmitter::emit()
7//!                         │
8//!              tokio::sync::broadcast (bounded, 1024)
9//!                         │
10//!           ┌─────────────┴──────────────┐
11//!      in-process                  TraceTransport::accept_loop
12//!      subscriber                  (UDS on Unix, TCP fallback)
13//!                                        │
14//!                                  up to 4 GUI connections
15//!                                  (newline-delimited JSON)
16//! ```
17//!
18//! # Transport variants
19//!
20//! | `TraceTransportConfig` | Platform | Notes |
21//! |---|---|---|
22//! | `Uds { path }` | Unix/macOS | Default when available |
23//! | `Tcp { addr }` | All | Portable fallback; `addr = "127.0.0.1:0"` assigns ephemeral port |
24//! | `Disabled` | All | No out-of-process forwarding (default) |
25//!
26//! # Back-pressure
27//!
28//! The broadcast channel is bounded by [`TRACE_CHANNEL_CAPACITY`]. When
29//! the channel is full, `emit` drops the event and increments an internal
30//! counter; the count is reported as `dropped_count` on the next event.
31//!
32//! Slow out-of-process subscribers receive a `RecvError::Lagged` from the
33//! broadcast channel; the gap is reported in the next JSON line via
34//! `dropped_count`.
35//!
36//! # Subscriber cap
37//!
38//! At most [`MAX_SUBSCRIBERS`] out-of-process connections are accepted.
39//! A fifth connection receives `{"error":"max_subscribers_reached"}` and
40//! is then closed.
41
42use std::sync::{
43    Arc,
44    atomic::{AtomicU32, AtomicUsize, Ordering},
45};
46use std::time::{Duration, SystemTime, UNIX_EPOCH};
47
48use serde::Serialize;
49use tokio::io::AsyncWriteExt;
50use tokio::sync::broadcast;
51
52/// Capacity of the broadcast channel (events).
53pub const TRACE_CHANNEL_CAPACITY: usize = 1_024;
54/// Maximum concurrent out-of-process subscriber connections.
55pub const MAX_SUBSCRIBERS: usize = 4;
56
57// ── Event schema ──────────────────────────────────────────────────────
58
59/// A single request/response trace event.
60#[derive(Clone, Debug, Serialize)]
61pub struct MatchTraceEvent {
62    /// Monotonically increasing event counter within this server run.
63    pub event_id: u64,
64    /// Schema version — bumped on breaking changes.
65    pub schema_version: u8,
66    /// Unix timestamp (milliseconds) when the request was received.
67    pub received_at_ms: u64,
68    /// Processing time in milliseconds.
69    pub duration_ms: u32,
70    /// Key fields from the incoming request.
71    pub request: RequestSummary,
72    /// What the server decided to do with the request.
73    pub outcome: Outcome,
74    /// Events dropped since the last successfully delivered event.
75    pub dropped_count: u32,
76}
77
78/// Key fields from the incoming HTTP request.
79#[derive(Clone, Debug, Serialize)]
80pub struct RequestSummary {
81    pub method: String,
82    pub url_path: String,
83    /// Selected request headers (display-only).
84    pub headers: Vec<(String, String)>,
85    /// Captured JSON body (RFC 023). Present only when `TraceConfig::capture_body`
86    /// is `true` and the request body is valid JSON within the size cap.
87    #[serde(skip_serializing_if = "Option::is_none")]
88    pub body_json: Option<serde_json::Value>,
89    /// `true` when the body was omitted because it exceeded `max_body_bytes`.
90    #[serde(skip_serializing_if = "std::ops::Not::not")]
91    pub body_truncated: bool,
92}
93
94impl RequestSummary {
95    /// Construct with no body capture (the common case before RFC 023).
96    pub fn without_body(method: String, url_path: String, headers: Vec<(String, String)>) -> Self {
97        Self { method, url_path, headers, body_json: None, body_truncated: false }
98    }
99}
100
101/// Trace-channel behaviour configuration (RFC 023).
102#[derive(Clone, Debug)]
103pub struct TraceConfig {
104    /// Capture the JSON request body in each event. Default: `false`.
105    pub capture_body: bool,
106    /// Maximum serialised body size in bytes. Bodies larger than this
107    /// are omitted and `body_truncated = true` is set. Default: 8 192.
108    pub max_body_bytes: usize,
109}
110
111impl Default for TraceConfig {
112    fn default() -> Self {
113        Self { capture_body: false, max_body_bytes: 8_192 }
114    }
115}
116
117/// What the server decided to do with the request.
118#[derive(Clone, Debug, Serialize)]
119#[serde(tag = "type", rename_all = "snake_case")]
120pub enum Outcome {
121    Matched { rule_set_index: usize, rule_index: usize },
122    Fallback { file_path: String, status: u16 },
123    Miss    { status: u16 },
124    Error   { kind: String, message: String },
125}
126
127// ── Emitter ───────────────────────────────────────────────────────────
128
129/// Shared handle to the trace broadcast channel.
130///
131/// Clone freely — each clone refers to the same underlying channel.
132#[derive(Clone)]
133pub struct TraceEmitter {
134    sender:          broadcast::Sender<MatchTraceEvent>,
135    event_counter:   Arc<AtomicU32>,
136    dropped_counter: Arc<AtomicU32>,
137    /// Behaviour settings (body capture, etc.).
138    pub config:      Arc<TraceConfig>,
139}
140
141impl TraceEmitter {
142    pub fn new() -> Self {
143        Self::with_config(TraceConfig::default())
144    }
145
146    pub fn with_config(config: TraceConfig) -> Self {
147        let (sender, _) = broadcast::channel(TRACE_CHANNEL_CAPACITY);
148        Self {
149            sender,
150            event_counter:   Arc::new(AtomicU32::new(0)),
151            dropped_counter: Arc::new(AtomicU32::new(0)),
152            config:          Arc::new(config),
153        }
154    }
155
156    /// Subscribe to the event stream (in-process).
157    pub fn subscribe(&self) -> broadcast::Receiver<MatchTraceEvent> {
158        self.sender.subscribe()
159    }
160
161    /// Attach body JSON to a `RequestSummary` according to this emitter's
162    /// `TraceConfig`. Call before `emit` when the request body is available.
163    pub fn enrich_with_body(
164        &self,
165        summary: &mut RequestSummary,
166        body_json: Option<&serde_json::Value>,
167    ) {
168        if !self.config.capture_body {
169            return;
170        }
171        match body_json {
172            None => {} // non-JSON or empty body — leave body_json = None
173            Some(v) => {
174                // Check serialised size against the cap.
175                match serde_json::to_string(v) {
176                    Ok(s) if s.len() <= self.config.max_body_bytes => {
177                        summary.body_json = Some(v.clone());
178                    }
179                    Ok(_) => {
180                        summary.body_truncated = true;
181                    }
182                    Err(_) => {} // shouldn't happen for a valid Value
183                }
184            }
185        }
186    }
187
188    /// Emit one event.  If the channel is full, the event is dropped and
189    /// the internal drop counter incremented.
190    pub fn emit(
191        &self,
192        received_at_ms: u64,
193        duration_ms: u32,
194        request: RequestSummary,
195        outcome: Outcome,
196    ) {
197        let event_id    = self.event_counter.fetch_add(1, Ordering::Relaxed) as u64;
198        let dropped_count = self.dropped_counter.swap(0, Ordering::Relaxed);
199
200        let event = MatchTraceEvent {
201            event_id,
202            schema_version: 1,
203            received_at_ms,
204            duration_ms,
205            request,
206            outcome,
207            dropped_count,
208        };
209
210        if self.sender.send(event).is_err() {
211            self.dropped_counter.fetch_add(1, Ordering::Relaxed);
212        }
213    }
214
215    /// Returns `true` iff at least one receiver is currently active.
216    pub fn has_subscribers(&self) -> bool {
217        self.sender.receiver_count() > 0
218    }
219}
220
221impl Default for TraceEmitter { fn default() -> Self { Self::new() } }
222
223// ── Transport configuration ───────────────────────────────────────────
224
225/// Configuration for the out-of-process transport layer.
226#[derive(Clone, Debug, Default)]
227pub enum TraceTransportConfig {
228    /// Unix-domain socket at the given path (Unix/macOS only).
229    #[cfg(unix)]
230    Uds { path: String },
231    /// TCP loopback socket (portable fallback).
232    Tcp { addr: String },
233    /// No out-of-process forwarding.
234    #[default]
235    Disabled,
236}
237
238// ── Transport implementation ──────────────────────────────────────────
239
240pub struct TraceTransport;
241
242impl TraceTransport {
243    /// Start accepting out-of-process subscriber connections and forwarding
244    /// events as newline-delimited JSON.
245    ///
246    /// This future runs forever (until the process exits or the socket
247    /// errors fatally). Spawn it with `tokio::spawn`.
248    ///
249    /// # Subscriber cap
250    ///
251    /// At most [`MAX_SUBSCRIBERS`] connections are served simultaneously.
252    /// Connection #`MAX_SUBSCRIBERS + 1` receives a JSON error line and
253    /// is closed.
254    pub async fn accept_loop(config: TraceTransportConfig, emitter: TraceEmitter) {
255        match config {
256            #[cfg(unix)]
257            TraceTransportConfig::Uds { path } => {
258                Self::uds_accept_loop(path, emitter).await
259            }
260            TraceTransportConfig::Tcp { addr } => {
261                Self::tcp_accept_loop(addr, emitter).await
262            }
263            TraceTransportConfig::Disabled => {
264                // No-op — transport disabled; in-process channel still works.
265            }
266        }
267    }
268
269    // ── TCP accept loop ───────────────────────────────────────────────
270
271    async fn tcp_accept_loop(addr: String, emitter: TraceEmitter) {
272        let listener = match tokio::net::TcpListener::bind(&addr).await {
273            Ok(l) => {
274                let bound = l.local_addr().map(|a| a.to_string())
275                    .unwrap_or_else(|_| addr.clone());
276                log::info!("trace transport: TCP listening on {}", bound);
277                l
278            }
279            Err(e) => {
280                log::error!("trace transport: failed to bind TCP {}: {}", addr, e);
281                return;
282            }
283        };
284
285        let active = Arc::new(AtomicUsize::new(0));
286        loop {
287            match listener.accept().await {
288                Ok((stream, peer)) => {
289                    let count = active.fetch_add(1, Ordering::Relaxed) + 1;
290                    if count > MAX_SUBSCRIBERS {
291                        active.fetch_sub(1, Ordering::Relaxed);
292                        let active_clone = active.clone();
293                        tokio::spawn(async move {
294                            let (_, mut writer) = tokio::io::split(stream);
295                            let _ = writer
296                                .write_all(b"{\"error\":\"max_subscribers_reached\"}\n")
297                                .await;
298                            drop(active_clone);
299                        });
300                        continue;
301                    }
302                    log::debug!("trace: TCP subscriber connected from {}", peer);
303                    let rx = emitter.subscribe();
304                    let active_clone = active.clone();
305                    tokio::spawn(async move {
306                        let (_, writer) = tokio::io::split(stream);
307                        Self::forward_events(writer, rx).await;
308                        active_clone.fetch_sub(1, Ordering::Relaxed);
309                        log::debug!("trace: TCP subscriber {} disconnected", peer);
310                    });
311                }
312                Err(e) => {
313                    log::error!("trace: TCP accept error: {}", e);
314                    tokio::time::sleep(Duration::from_millis(100)).await;
315                }
316            }
317        }
318    }
319
320    // ── UDS accept loop (Unix only) ───────────────────────────────────
321
322    #[cfg(unix)]
323    async fn uds_accept_loop(path: String, emitter: TraceEmitter) {
324        // Remove stale socket file from a previous run.
325        let _ = std::fs::remove_file(&path);
326
327        let listener = match tokio::net::UnixListener::bind(&path) {
328            Ok(l) => {
329                log::info!("trace transport: UDS listening at {}", path);
330                l
331            }
332            Err(e) => {
333                log::error!("trace transport: failed to bind UDS {}: {}", path, e);
334                return;
335            }
336        };
337
338        let active = Arc::new(AtomicUsize::new(0));
339        loop {
340            match listener.accept().await {
341                Ok((stream, _)) => {
342                    let count = active.fetch_add(1, Ordering::Relaxed) + 1;
343                    if count > MAX_SUBSCRIBERS {
344                        active.fetch_sub(1, Ordering::Relaxed);
345                        tokio::spawn(async move {
346                            let (_, mut writer) = tokio::io::split(stream);
347                            let _ = writer
348                                .write_all(b"{\"error\":\"max_subscribers_reached\"}\n")
349                                .await;
350                        });
351                        continue;
352                    }
353                    log::debug!("trace: UDS subscriber connected");
354                    let rx = emitter.subscribe();
355                    let active_clone = active.clone();
356                    tokio::spawn(async move {
357                        let (_, writer) = tokio::io::split(stream);
358                        Self::forward_events(writer, rx).await;
359                        active_clone.fetch_sub(1, Ordering::Relaxed);
360                        log::debug!("trace: UDS subscriber disconnected");
361                    });
362                }
363                Err(e) => {
364                    log::error!("trace: UDS accept error: {}", e);
365                    tokio::time::sleep(Duration::from_millis(100)).await;
366                }
367            }
368        }
369    }
370
371    // ── Event forwarder (shared by UDS and TCP) ───────────────────────
372
373    /// Read events from `rx` and write each as a JSON line to `writer`
374    /// until the connection closes or the channel is closed.
375    async fn forward_events<W>(mut writer: W, mut rx: broadcast::Receiver<MatchTraceEvent>)
376    where
377        W: tokio::io::AsyncWrite + Unpin,
378    {
379        loop {
380            let event = match rx.recv().await {
381                Ok(e) => e,
382                Err(broadcast::error::RecvError::Lagged(n)) => {
383                    // Receiver was too slow; `n` events were dropped.
384                    // The next event will carry `dropped_count` so the
385                    // subscriber can detect the gap. Continue.
386                    log::debug!("trace: subscriber lagged, {} events dropped", n);
387                    continue;
388                }
389                Err(broadcast::error::RecvError::Closed) => break,
390            };
391
392            let mut line = match serde_json::to_string(&event) {
393                Ok(s) => s,
394                Err(e) => {
395                    log::error!("trace: serialise error: {}", e);
396                    continue;
397                }
398            };
399            line.push('\n');
400
401            if writer.write_all(line.as_bytes()).await.is_err() {
402                break; // subscriber disconnected
403            }
404        }
405    }
406}
407
408// ── Timestamp helper ──────────────────────────────────────────────────
409
410/// Current Unix time in milliseconds.
411pub fn now_ms() -> u64 {
412    SystemTime::now()
413        .duration_since(UNIX_EPOCH)
414        .unwrap_or(Duration::ZERO)
415        .as_millis() as u64
416}
417
418// ── Tests ─────────────────────────────────────────────────────────────
419
420#[cfg(test)]
421mod tests {
422    use super::*;
423
424    #[tokio::test]
425    async fn emit_received_by_subscriber() {
426        let emitter = TraceEmitter::new();
427        let mut rx = emitter.subscribe();
428
429        emitter.emit(
430            1_000_000, 5,
431            RequestSummary { method: "GET".into(), url_path: "/api/test".into(), headers: vec![], body_json: None, body_truncated: false },
432            Outcome::Miss { status: 404 },
433        );
434
435        let event = rx.try_recv().expect("event in channel");
436        assert_eq!(event.event_id, 0);
437        assert_eq!(event.schema_version, 1);
438        assert_eq!(event.request.method, "GET");
439        assert_eq!(event.duration_ms, 5);
440        assert_eq!(event.dropped_count, 0);
441        assert!(matches!(event.outcome, Outcome::Miss { status: 404 }));
442    }
443
444    #[tokio::test]
445    async fn emit_no_subscriber_increments_dropped() {
446        let emitter = TraceEmitter::new();
447        emitter.emit(0, 0,
448            RequestSummary { method: "GET".into(), url_path: "/".into(), headers: vec![], body_json: None, body_truncated: false },
449            Outcome::Miss { status: 404 },
450        );
451        let mut rx = emitter.subscribe();
452        emitter.emit(0, 0,
453            RequestSummary { method: "GET".into(), url_path: "/".into(), headers: vec![], body_json: None, body_truncated: false },
454            Outcome::Miss { status: 200 },
455        );
456        let event = rx.try_recv().expect("second event visible");
457        assert_eq!(event.dropped_count, 1, "first event should be counted dropped");
458    }
459
460    #[test]
461    fn has_subscribers_reflects_state() {
462        let emitter = TraceEmitter::new();
463        assert!(!emitter.has_subscribers());
464        let _rx = emitter.subscribe();
465        assert!(emitter.has_subscribers());
466    }
467
468    #[tokio::test]
469    async fn outcome_serialises_correctly() {
470        let event = MatchTraceEvent {
471            event_id: 7, schema_version: 1, received_at_ms: 0, duration_ms: 0,
472            request: RequestSummary { method: "POST".into(), url_path: "/x".into(), headers: vec![], body_json: None, body_truncated: false },
473            outcome: Outcome::Matched { rule_set_index: 0, rule_index: 2 },
474            dropped_count: 0,
475        };
476        let json = serde_json::to_string(&event).unwrap();
477        assert!(json.contains("\"type\":\"matched\""));
478        assert!(json.contains("\"rule_index\":2"));
479        assert!(json.contains("\"schema_version\":1"));
480    }
481
482    #[tokio::test]
483    async fn tcp_transport_delivers_events() {
484        let emitter = TraceEmitter::new();
485        let emitter_clone = emitter.clone();
486
487        // Bind on an ephemeral port.
488        let config = TraceTransportConfig::Tcp { addr: "127.0.0.1:0".to_owned() };
489
490        // We need to know the actual bound port before connecting.
491        // Bind the listener ourselves to capture the address, then hand
492        // the address to the transport accept loop via a channel.
493        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
494        let bound_addr = listener.local_addr().unwrap();
495
496        // Spawn a simplified accept loop that uses our pre-bound listener.
497        tokio::spawn(async move {
498            let (stream, _) = listener.accept().await.unwrap();
499            let rx = emitter_clone.subscribe();
500            let (_, writer) = tokio::io::split(stream);
501            TraceTransport::forward_events(writer, rx).await;
502        });
503
504        // Connect a subscriber.
505        let mut client = tokio::net::TcpStream::connect(bound_addr).await.unwrap();
506
507        // Give the subscriber task a moment to subscribe before emitting.
508        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
509
510        emitter.emit(
511            42, 3,
512            RequestSummary { method: "GET".into(), url_path: "/ping".into(), headers: vec![], body_json: None, body_truncated: false },
513            Outcome::Miss { status: 404 },
514        );
515
516        // Read one JSON line from the TCP connection.
517        use tokio::io::AsyncBufReadExt;
518        let mut reader = tokio::io::BufReader::new(&mut client);
519        let mut line = String::new();
520        tokio::time::timeout(
521            std::time::Duration::from_secs(2),
522            reader.read_line(&mut line),
523        )
524        .await
525        .expect("timeout")
526        .expect("read ok");
527
528        let value: serde_json::Value = serde_json::from_str(line.trim()).expect("valid JSON");
529        assert_eq!(value["request"]["url_path"], "/ping");
530        assert_eq!(value["outcome"]["type"], "miss");
531        assert_eq!(value["schema_version"], 1);
532    }
533
534    // ── RFC 023: body capture tests ───────────────────────────────────
535
536    #[test]
537    fn enrich_with_body_disabled_by_default() {
538        let emitter = TraceEmitter::new(); // capture_body = false by default
539        let mut summary = RequestSummary {
540            method: "POST".into(), url_path: "/".into(), headers: vec![],
541            body_json: None, body_truncated: false,
542        };
543        let body = serde_json::json!({"action": "create"});
544        emitter.enrich_with_body(&mut summary, Some(&body));
545        assert!(summary.body_json.is_none(), "body should not be captured when disabled");
546        assert!(!summary.body_truncated);
547    }
548
549    #[test]
550    fn enrich_with_body_enabled_captures_small_body() {
551        let emitter = TraceEmitter::with_config(TraceConfig { capture_body: true, max_body_bytes: 8_192 });
552        let mut summary = RequestSummary {
553            method: "POST".into(), url_path: "/".into(), headers: vec![],
554            body_json: None, body_truncated: false,
555        };
556        let body = serde_json::json!({"action": "create", "user_id": 42});
557        emitter.enrich_with_body(&mut summary, Some(&body));
558        assert!(summary.body_json.is_some(), "body should be captured when enabled");
559        assert_eq!(summary.body_json.unwrap()["action"], "create");
560        assert!(!summary.body_truncated);
561    }
562
563    #[test]
564    fn enrich_with_body_truncates_oversized_body() {
565        let emitter = TraceEmitter::with_config(TraceConfig { capture_body: true, max_body_bytes: 10 });
566        let mut summary = RequestSummary {
567            method: "POST".into(), url_path: "/".into(), headers: vec![],
568            body_json: None, body_truncated: false,
569        };
570        let body = serde_json::json!({"data": "this is longer than 10 bytes"});
571        emitter.enrich_with_body(&mut summary, Some(&body));
572        assert!(summary.body_json.is_none(), "oversized body should be omitted");
573        assert!(summary.body_truncated, "body_truncated flag should be set");
574    }
575
576    #[test]
577    fn request_summary_body_json_not_in_serialised_output_when_none() {
578        let summary = RequestSummary {
579            method: "GET".into(), url_path: "/api".into(), headers: vec![],
580            body_json: None, body_truncated: false,
581        };
582        let json = serde_json::to_string(&summary).unwrap();
583        assert!(!json.contains("body_json"), "absent body_json must be skipped");
584        assert!(!json.contains("body_truncated"), "false body_truncated must be skipped");
585    }
586}
apimock_server/trace.rs

apimock_server/
trace.rs
