saddle_observability/
call.rs

1use std::time::{Instant, SystemTime, UNIX_EPOCH};
2
3use saddle_core::{
4    ApplicationId, CallContext, ErrorKind, ModuleId, OperationId, SaddleError, ServiceId, SpanId,
5    TraceCorrelationId,
6};
7use serde_json::{Value, json};
8use serde::Serialize;
9
10use crate::{
11    DomainEvent, EventLevel, InboundTrace, Observer, TraceIdError,
12    logger::LogRecord,
13    trace::{select_entry_trace_id, select_trace_id},
14};
15
16/// Framework boundaries that automatically produce start and finish records.
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18pub enum CallKind {
19    Decode,
20    Admission,
21    RuntimeDispatch,
22    Handler,
23    ExternalRequest,
24    Service,
25    Database,
26    Transaction,
27    ExternalDecision,
28    ExternalConnect,
29    ExternalUnary,
30    ExternalCall,
31    Finalizer,
32    Response,
33}
34
35impl CallKind {
36    const fn as_str(self) -> &'static str {
37        match self {
38            Self::Decode => "decode",
39            Self::Admission => "admission",
40            Self::RuntimeDispatch => "runtime_dispatch",
41            Self::Handler => "handler",
42            Self::ExternalRequest => "external_request",
43            Self::Service => "service",
44            Self::Database => "database",
45            Self::Transaction => "transaction",
46            Self::ExternalDecision => "external_decision",
47            Self::ExternalConnect => "external_connect",
48            Self::ExternalUnary => "external_unary",
49            Self::ExternalCall => "external_call",
50            Self::Finalizer => "finalizer",
51            Self::Response => "response",
52        }
53    }
54}
55
56/// Stable completion classification for framework call logs.
57#[derive(Clone, Copy, Debug, Eq, PartialEq)]
58pub enum CallOutcome {
59    Success,
60    Failure,
61    Abandoned,
62}
63
64impl CallOutcome {
65    const fn as_str(self) -> &'static str {
66        match self {
67            Self::Success => "success",
68            Self::Failure => "failure",
69            Self::Abandoned => "abandoned",
70        }
71    }
72}
73
74/// An in-flight framework boundary.
75///
76/// Dropping it without calling [`ActiveCall::succeed`] or
77/// [`ActiveCall::fail`] emits an `abandoned` finish record, which preserves a
78/// balanced trace when a future is cancelled or unwinds.
79pub struct ActiveCall {
80    observer: Observer,
81    context: CallContext,
82    parent_span_id: Option<SpanId>,
83    kind: CallKind,
84    started_at: Instant,
85    finished: bool,
86    process_log: bool,
87}
88
89#[derive(Clone, Copy)]
90struct SpanText(SpanId);
91
92impl Serialize for SpanText {
93    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
94        const HEX: &[u8; 16] = b"0123456789abcdef";
95        let mut bytes = [0_u8; 16];
96        let mut value = self.0.as_u64();
97        for index in (0..16).rev() {
98            bytes[index] = HEX[(value & 15) as usize];
99            value >>= 4;
100        }
101        serializer.serialize_str(std::str::from_utf8(&bytes).expect("hex is utf8"))
102    }
103}
104
105// Borrowed schema for formal routed RPC call events. Serialization counts
106// bytes without allocating, then writes directly into one prepaid block.
107#[derive(Serialize)]
108pub(crate) struct ProcessCallRecord<'a> {
109    timestamp_unix_ms: u128,
110    level: EventLevel,
111    event: &'static str,
112    trace_id: &'a str,
113    span: &'static str,
114    span_id: SpanText,
115    #[serde(skip_serializing_if = "Option::is_none")]
116    parent: Option<SpanText>,
117    #[serde(skip_serializing_if = "Option::is_none")]
118    parent_span_id: Option<SpanText>,
119    rpc_id: Option<&'a str>,
120    request: &'a str,
121    call: &'static str,
122    call_kind: &'static str,
123    app: &'a str,
124    application: &'a str,
125    zone: &'a str,
126    interface: &'a str,
127    module: &'a str,
128    service: &'a str,
129    operation: &'a str,
130    #[serde(skip_serializing_if = "Option::is_none")]
131    trace_source: Option<&'static str>,
132    #[serde(skip_serializing_if = "Option::is_none")]
133    outcome: Option<&'static str>,
134    #[serde(skip_serializing_if = "Option::is_none")]
135    duration: Option<u64>,
136    #[serde(skip_serializing_if = "Option::is_none")]
137    elapsed_us: Option<u64>,
138    #[serde(skip_serializing_if = "Option::is_none")]
139    error_kind: Option<&'static str>,
140    #[serde(skip_serializing_if = "Option::is_none")]
141    error_code: Option<&'a str>,
142    #[serde(skip_serializing_if = "call_zero")]
143    pub(crate) dropped_events: u64,
144}
145
146const fn call_zero(value: &u64) -> bool { *value == 0 }
147
148impl<'a> ProcessCallRecord<'a> {
149    fn new(
150        context: &'a CallContext, parent: Option<SpanId>, kind: CallKind,
151        level: EventLevel, event: &'static str, trace_source: Option<&'static str>,
152        outcome: Option<(&'static str, u64)>, error: Option<&'a SaddleError>,
153    ) -> Self {
154        let app = context.application().as_str();
155        let zone = context.module().as_str();
156        let interface = context.service().as_str();
157        let operation = context.operation().as_str();
158        Self {
159            timestamp_unix_ms: SystemTime::now().duration_since(UNIX_EPOCH)
160                .unwrap_or_default().as_millis(),
161            level, event,
162            trace_id: context.trace_correlation_id().as_str(),
163            span: kind.as_str(), span_id: SpanText(context.span_id()),
164            parent: parent.map(SpanText), parent_span_id: parent.map(SpanText),
165            rpc_id: context.rpc_correlation_id().map(|rpc| rpc.as_str()),
166            request: operation, call: kind.as_str(), call_kind: kind.as_str(),
167            app, application: app, zone, interface, module: zone,
168            service: interface, operation, trace_source,
169            outcome: outcome.map(|(name, _)| name),
170            duration: outcome.map(|(_, duration)| duration),
171            elapsed_us: outcome.map(|(_, duration)| duration),
172            error_kind: error.map(|error| error_kind(error.kind())),
173            error_code: error.map(SaddleError::code),
174            dropped_events: 0,
175        }
176    }
177}
178
179struct SelectedEntryTrace {
180    internal: saddle_core::TraceId,
181    correlation: Option<TraceCorrelationId>,
182    source: InboundTrace,
183    rpc: Option<saddle_core::RpcCorrelationId>,
184}
185
186impl Observer {
187    /// Starts an external request using the legacy compatibility policy.
188    ///
189    /// Invalid supplied identifiers are replaced for compatibility with 0.3.0
190    /// callers. Production entry points must use
191    /// [`Observer::start_external_call_checked`] so only a missing identifier
192    /// can cause generation.
193    pub fn start_external_call(
194        &self,
195        application: impl Into<ApplicationId>,
196        module: impl Into<ModuleId>,
197        service: impl Into<ServiceId>,
198        operation: impl Into<OperationId>,
199        inbound_trace_id: Option<&str>,
200    ) -> (ActiveCall, InboundTrace) {
201        let (trace_id, inbound) = select_trace_id(inbound_trace_id, || self.new_trace_id());
202        self.start_selected_external_call(
203            application,
204            module,
205            service,
206            operation,
207            SelectedEntryTrace {
208                internal: trace_id,
209                correlation: None,
210                source: inbound,
211                rpc: None,
212            },
213            false,
214        )
215    }
216
217    /// Starts a production entry call. A supplied opaque protocol identifier
218    /// is inherited byte-for-byte after bounded validation; only an absent
219    /// identifier may cause generation.
220    pub fn start_external_call_checked(
221        &self,
222        application: impl Into<ApplicationId>,
223        module: impl Into<ModuleId>,
224        service: impl Into<ServiceId>,
225        operation: impl Into<OperationId>,
226        inbound_trace_id: Option<&str>,
227    ) -> Result<(ActiveCall, InboundTrace), TraceIdError> {
228        let (trace_id, trace_correlation_id, inbound) =
229            select_entry_trace_id(inbound_trace_id, || self.new_trace_id())?;
230        Ok(self.start_selected_external_call(
231            application,
232            module,
233            service,
234            operation,
235            SelectedEntryTrace {
236                internal: trace_id,
237                correlation: Some(trace_correlation_id),
238                source: inbound,
239                rpc: None,
240            },
241            false,
242        ))
243    }
244
245    /// Entry supplied protocol RPC, already validated by its typed carrier.
246    /// No child RPC is created for local stages or database transactions.
247    pub fn start_external_call_with_rpc(
248        &self,
249        application: impl Into<ApplicationId>,
250        module: impl Into<ModuleId>,
251        service: impl Into<ServiceId>,
252        operation: impl Into<OperationId>,
253        inbound_trace_id: Option<&str>,
254        rpc: saddle_core::RpcCorrelationId,
255    ) -> Result<(ActiveCall, InboundTrace), TraceIdError> {
256        let (internal, correlation, source) =
257            select_entry_trace_id(inbound_trace_id, || self.new_trace_id())?;
258        Ok(self.start_selected_external_call(
259            application,
260            module,
261            service,
262            operation,
263            SelectedEntryTrace {
264                internal,
265                correlation: Some(correlation),
266                source,
267                rpc: Some(rpc),
268            },
269            false,
270        ))
271    }
272
273    /// Only the guarded routed RPC uses the prepaid process-owned call log.
274    #[doc(hidden)]
275    pub fn start_managed_external_call_with_rpc(
276        &self,
277        application: impl Into<ApplicationId>,
278        module: impl Into<ModuleId>,
279        service: impl Into<ServiceId>,
280        operation: impl Into<OperationId>,
281        inbound_trace_id: Option<&str>,
282        rpc: saddle_core::RpcCorrelationId,
283    ) -> Result<(ActiveCall, InboundTrace), TraceIdError> {
284        let (internal, correlation, source) =
285            select_entry_trace_id(inbound_trace_id, || self.new_trace_id())?;
286        Ok(self.start_selected_external_call(application, module, service, operation,
287            SelectedEntryTrace { internal, correlation: Some(correlation), source, rpc: Some(rpc) }, true))
288    }
289
290    fn start_selected_external_call(
291        &self,
292        application: impl Into<ApplicationId>,
293        module: impl Into<ModuleId>,
294        service: impl Into<ServiceId>,
295        operation: impl Into<OperationId>,
296        trace: SelectedEntryTrace,
297        process_log: bool,
298    ) -> (ActiveCall, InboundTrace) {
299        let mut context = CallContext::new(
300            application.into(),
301            module.into(),
302            service.into(),
303            operation.into(),
304            trace.internal,
305            self.new_span_id(),
306        )
307        .with_rpc_correlation_id(trace.rpc);
308        if let Some(trace_correlation_id) = trace.correlation {
309            context = context.with_trace_correlation_id(trace_correlation_id);
310        }
311        (
312            ActiveCall::start(
313                self.clone(),
314                context,
315                None,
316                CallKind::ExternalRequest,
317                Some(trace.source),
318                process_log,
319            ),
320            trace.source,
321        )
322    }
323
324    /// Starts a Service, DB or transaction child call on the parent's trace.
325    pub fn start_child_call(
326        &self,
327        parent: &CallContext,
328        kind: CallKind,
329        module: impl Into<ModuleId>,
330        service: impl Into<ServiceId>,
331        operation: impl Into<OperationId>,
332    ) -> ActiveCall {
333        let context = CallContext::new(
334            parent.application().clone(),
335            module.into(),
336            service.into(),
337            operation.into(),
338            parent.trace_id(),
339            self.new_span_id(),
340        )
341        .with_trace_correlation_id(parent.trace_correlation_id().clone())
342        .with_rpc_correlation_id(parent.rpc_correlation_id().cloned());
343        ActiveCall::start(self.clone(), context, Some(parent.span_id()), kind, None, false)
344    }
345
346    /// Child identity for a guarded source stage; immutable identifiers have
347    /// already been allocated and frozen by application startup.
348    #[doc(hidden)]
349    pub fn managed_child_context(&self, parent: &CallContext, module: ModuleId,
350        service: ServiceId, operation: OperationId) -> CallContext {
351        CallContext::new(parent.application().clone(), module, service, operation,
352            parent.trace_id(), self.new_span_id())
353            .with_trace_correlation_id(parent.trace_correlation_id().clone())
354            .with_rpc_correlation_id(parent.rpc_correlation_id().cloned())
355    }
356
357    #[doc(hidden)]
358    pub fn start_managed_child_call(
359        &self, parent: &CallContext, kind: CallKind,
360        module: impl Into<ModuleId>, service: impl Into<ServiceId>,
361        operation: impl Into<OperationId>,
362    ) -> ActiveCall {
363        let context = CallContext::new(parent.application().clone(), module.into(), service.into(),
364            operation.into(), parent.trace_id(), self.new_span_id())
365            .with_trace_correlation_id(parent.trace_correlation_id().clone())
366            .with_rpc_correlation_id(parent.rpc_correlation_id().cloned());
367        ActiveCall::start(self.clone(), context, Some(parent.span_id()), kind, None, true)
368    }
369
370    /// Records an intentional domain event without accepting arbitrary objects.
371    pub fn record_domain_event(&self, context: &CallContext, event: DomainEvent) {
372        let mut record = LogRecord::new(event.level, "domain.event");
373        add_context(&mut record, context, None);
374        add_identity(&mut record, context);
375        record
376            .data
377            .insert("name".to_owned(), Value::String(event.name));
378        let fields = event
379            .fields
380            .into_iter()
381            .map(|(name, value)| (name, value.into_json()))
382            .collect();
383        record
384            .data
385            .insert("fields".to_owned(), Value::Object(fields));
386        self.emit(record);
387    }
388}
389
390impl ActiveCall {
391    fn start(
392        observer: Observer,
393        context: CallContext,
394        parent_span_id: Option<SpanId>,
395        kind: CallKind,
396        inbound_trace: Option<InboundTrace>,
397        process_log: bool,
398    ) -> Self {
399        if process_log {
400            let mut record = ProcessCallRecord::new(&context, parent_span_id, kind,
401                EventLevel::Info, "framework.call.started", inbound_trace.map(InboundTrace::as_str),
402                None, None);
403            observer.emit_process_call(&mut record);
404        } else {
405        let mut record = LogRecord::new(EventLevel::Info, "framework.call.started");
406        add_context(&mut record, &context, parent_span_id);
407        add_call_identity(&mut record, &context, kind);
408        if let Some(inbound_trace) = inbound_trace {
409            record.data.insert(
410                "trace_source".to_owned(),
411                Value::String(inbound_trace.as_str().to_owned()),
412            );
413        }
414        observer.emit(record);
415        }
416
417        Self {
418            observer,
419            context,
420            parent_span_id,
421            kind,
422            started_at: Instant::now(),
423            finished: false,
424            process_log,
425        }
426    }
427
428    pub fn context(&self) -> &CallContext {
429        &self.context
430    }
431
432    pub fn succeed(mut self) {
433        self.finish(CallOutcome::Success, None);
434    }
435
436    pub fn fail(mut self, error: &SaddleError) {
437        self.finish(CallOutcome::Failure, Some(error));
438    }
439
440    fn finish(&mut self, outcome: CallOutcome, error: Option<&SaddleError>) {
441        if self.finished {
442            return;
443        }
444
445        let level = match outcome {
446            CallOutcome::Success => EventLevel::Info,
447            CallOutcome::Failure | CallOutcome::Abandoned => EventLevel::Error,
448        };
449        if self.process_log {
450            let duration = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX);
451            let mut record = ProcessCallRecord::new(&self.context, self.parent_span_id, self.kind,
452                level, "framework.call.finished", None, Some((outcome.as_str(), duration)), error);
453            self.observer.emit_process_call(&mut record);
454            self.finished = true;
455            return;
456        }
457        let mut record = LogRecord::new(level, "framework.call.finished");
458        add_context(&mut record, &self.context, self.parent_span_id);
459        add_call_identity(&mut record, &self.context, self.kind);
460        record.data.insert(
461            "outcome".to_owned(),
462            Value::String(outcome.as_str().to_owned()),
463        );
464        let duration = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX);
465        record.data.insert("duration".to_owned(), json!(duration));
466        record.data.insert("elapsed_us".to_owned(), json!(duration));
467        if let Some(error) = error {
468            record.data.insert(
469                "error_kind".to_owned(),
470                Value::String(error_kind(error.kind()).to_owned()),
471            );
472            record.data.insert(
473                "error_code".to_owned(),
474                Value::String(error.code().to_owned()),
475            );
476        }
477        self.observer.emit(record);
478        self.finished = true;
479    }
480}
481
482impl Drop for ActiveCall {
483    fn drop(&mut self) {
484        self.finish(CallOutcome::Abandoned, None);
485    }
486}
487
488fn add_context(record: &mut LogRecord, context: &CallContext, parent_span_id: Option<SpanId>) {
489    record.trace_id = Some(context.trace_correlation_id().to_string());
490    record.span_id = Some(context.span_id().to_string());
491    record.parent = parent_span_id.map(|span_id| span_id.to_string());
492    record.parent_span_id = parent_span_id.map(|span_id| span_id.to_string());
493    record.data.insert(
494        "rpc_id".to_owned(),
495        context
496            .rpc_correlation_id()
497            .map(|rpc| Value::String(rpc.as_str().into()))
498            .unwrap_or(Value::Null),
499    );
500    record.data.insert(
501        "request".to_owned(),
502        Value::String(context.operation().to_string()),
503    );
504}
505
506fn add_call_identity(record: &mut LogRecord, context: &CallContext, kind: CallKind) {
507    record.span = Some(kind.as_str().to_owned());
508    record
509        .data
510        .insert("call".to_owned(), Value::String(kind.as_str().to_owned()));
511    record.data.insert(
512        "call_kind".to_owned(),
513        Value::String(kind.as_str().to_owned()),
514    );
515    add_identity(record, context);
516}
517
518fn add_identity(record: &mut LogRecord, context: &CallContext) {
519    record.data.insert(
520        "app".to_owned(),
521        Value::String(context.application().to_string()),
522    );
523    record.data.insert(
524        "application".to_owned(),
525        Value::String(context.application().to_string()),
526    );
527    record.data.insert(
528        "zone".to_owned(),
529        Value::String(context.module().to_string()),
530    );
531    record.data.insert(
532        "interface".to_owned(),
533        Value::String(context.service().to_string()),
534    );
535    record.data.insert(
536        "module".to_owned(),
537        Value::String(context.module().to_string()),
538    );
539    record.data.insert(
540        "service".to_owned(),
541        Value::String(context.service().to_string()),
542    );
543    record.data.insert(
544        "operation".to_owned(),
545        Value::String(context.operation().to_string()),
546    );
547}
548
549const fn error_kind(kind: ErrorKind) -> &'static str {
550    match kind {
551        ErrorKind::InvalidArgument => "invalid_argument",
552        ErrorKind::NotFound => "not_found",
553        ErrorKind::Conflict => "conflict",
554        ErrorKind::Business => "business",
555        ErrorKind::Unavailable => "unavailable",
556        ErrorKind::Infrastructure => "infrastructure",
557        ErrorKind::Internal => "internal",
558        _ => "unknown",
559    }
560}
561
562#[cfg(test)]
563mod tests {
564    use std::{
565        future::Future,
566        io,
567        sync::{Arc, Mutex},
568        task::{Context, Poll, Wake, Waker},
569        thread,
570    };
571
572    use saddle_core::{ErrorKind, SaddleError};
573
574    use super::*;
575    use crate::ObserverConfig;
576
577    struct ThreadWaker(thread::Thread);
578
579    impl Wake for ThreadWaker {
580        fn wake(self: Arc<Self>) {
581            self.0.unpark();
582        }
583    }
584
585    fn block_on<T>(future: impl Future<Output = T>) -> T {
586        let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
587        let mut context = Context::from_waker(&waker);
588        let mut future = std::pin::pin!(future);
589        loop {
590            match future.as_mut().poll(&mut context) {
591                Poll::Ready(output) => return output,
592                Poll::Pending => thread::park(),
593            }
594        }
595    }
596
597    #[derive(Clone, Default)]
598    struct Capture(Arc<Mutex<Vec<u8>>>);
599
600    impl io::Write for Capture {
601        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
602            self.0.lock().unwrap().extend_from_slice(bytes);
603            Ok(bytes.len())
604        }
605
606        fn flush(&mut self) -> io::Result<()> {
607            Ok(())
608        }
609    }
610
611    fn records(capture: &Capture) -> Vec<Value> {
612        String::from_utf8(capture.0.lock().unwrap().clone())
613            .unwrap()
614            .lines()
615            .map(|line| serde_json::from_str(line).unwrap())
616            .collect()
617    }
618
619    #[test]
620    fn external_and_child_calls_share_trace_and_record_parentage() {
621        let capture = Capture::default();
622        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
623        let trace = "00112233445566778899aabbccddeeff";
624        let (external, source) =
625            observer.start_external_call("shop", "orders", "orders", "create", Some(trace));
626        assert_eq!(source, InboundTrace::Inherited);
627        assert_eq!(external.context().trace_id().to_string(), trace);
628
629        let child = observer.start_child_call(
630            external.context(),
631            CallKind::Service,
632            "users",
633            "users",
634            "validate",
635        );
636        assert_eq!(child.context().trace_id(), external.context().trace_id());
637        assert!(child.context().rpc_correlation_id().is_none());
638        assert_ne!(child.context().span_id(), external.context().span_id());
639        child.succeed();
640        external.succeed();
641        block_on(observer.flush()).unwrap();
642
643        let records = records(&capture);
644        assert_eq!(records.len(), 4);
645        assert_eq!(records[1]["parent_span_id"], records[0]["span_id"]);
646        assert!(records.iter().all(|record| record["trace_id"] == trace));
647    }
648
649    #[test]
650    fn missing_protocol_rpc_is_null_never_internal_span() {
651        let capture = Capture::default();
652        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
653        let (call, _) = observer
654            .start_external_call_checked("shop", "entry", "orders", "create", None)
655            .unwrap();
656        observer
657            .start_child_call(call.context(), CallKind::Database, "db", "db", "query")
658            .succeed();
659        call.succeed();
660        block_on(observer.flush()).unwrap();
661        for record in records(&capture) {
662            assert!(record["rpc_id"].is_null());
663            assert!(record["span_id"].as_str().is_some());
664        }
665    }
666
667    #[test]
668    fn full_stage_chain_preserves_protocol_rpc_and_independent_spans() {
669        let capture = Capture::default();
670        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
671        let trace = "00112233445566778899aabbccddeeff";
672        let (root, _) = observer
673            .start_external_call_with_rpc(
674                "shop",
675                "entry",
676                "orders",
677                "create",
678                Some(trace),
679                saddle_core::RpcCorrelationId::new("0").unwrap(),
680            )
681            .unwrap();
682        let stages = [
683            CallKind::Decode,
684            CallKind::Admission,
685            CallKind::RuntimeDispatch,
686            CallKind::Handler,
687            CallKind::Database,
688            CallKind::ExternalDecision,
689            CallKind::Finalizer,
690            CallKind::Response,
691        ];
692        for stage in stages {
693            observer
694                .start_child_call(root.context(), stage, "framework", "orders", "create")
695                .succeed();
696        }
697        observer
698            .start_child_call(
699                root.context(),
700                CallKind::ExternalCall,
701                "rpc",
702                "inventory",
703                "reserve",
704            )
705            .succeed();
706        observer
707            .start_child_call(
708                root.context(),
709                CallKind::ExternalCall,
710                "rpc",
711                "payment",
712                "authorize",
713            )
714            .succeed();
715        root.succeed();
716        block_on(observer.flush()).unwrap();
717
718        let records = records(&capture);
719        let started: Vec<_> = records
720            .iter()
721            .filter(|record| record["event"] == "framework.call.started")
722            .collect();
723        assert_eq!(started.len(), 11);
724        assert!(started.iter().all(|record| record["trace_id"] == trace));
725        for record in &started {
726            for field in [
727                "trace_id",
728                "rpc_id",
729                "span",
730                "request",
731                "call",
732                "app",
733                "interface",
734                "zone",
735            ] {
736                assert!(record.get(field).is_some(), "missing {field}");
737            }
738            let serialized = serde_json::to_string(record).unwrap();
739            for forbidden in [
740                "payload",
741                "cookie",
742                "session",
743                "requestCtx",
744                "user_id",
745                "password",
746                "secret",
747                "db_value",
748            ] {
749                assert!(!serialized.contains(forbidden));
750            }
751        }
752        let external_span_ids: Vec<_> = started
753            .iter()
754            .filter(|record| record["span"] == "external_call")
755            .map(|record| record["span_id"].as_str().unwrap())
756            .collect();
757        assert_eq!(external_span_ids.len(), 2);
758        assert_ne!(external_span_ids[0], external_span_ids[1]);
759        assert!(started.iter().all(|record| record["rpc_id"] == "0"));
760        assert!(started.iter().skip(1).all(|record| {
761            record["parent"] == started[0]["span_id"] && record["span_id"] != started[0]["span_id"]
762        }));
763        assert!(
764            records
765                .iter()
766                .filter(|record| record["event"] == "framework.call.finished")
767                .all(|record| record.get("outcome").is_some() && record.get("duration").is_some())
768        );
769    }
770
771    #[test]
772    fn invalid_inbound_trace_is_rejected_without_generating_a_replacement() {
773        let capture = Capture::default();
774        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
775        let result = observer.start_external_call_checked(
776            "shop",
777            "orders",
778            "orders",
779            "create",
780            Some("not-a-valid-trace\n"),
781        );
782        assert!(matches!(result, Err(TraceIdError::ControlCharacter)));
783        block_on(observer.flush()).unwrap();
784        assert!(records(&capture).is_empty());
785    }
786
787    #[test]
788    fn failure_logs_classification_but_not_sensitive_message() {
789        let capture = Capture::default();
790        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
791        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
792        let error = SaddleError::new(
793            ErrorKind::Infrastructure,
794            "db.query_failed",
795            "password=must-not-be-logged",
796        );
797        call.fail(&error);
798        block_on(observer.flush()).unwrap();
799
800        let output = String::from_utf8(capture.0.lock().unwrap().clone()).unwrap();
801        assert!(output.contains("db.query_failed"));
802        assert!(output.contains("infrastructure"));
803        assert!(!output.contains("must-not-be-logged"));
804    }
805
806    #[test]
807    fn domain_event_contains_only_explicit_scalar_fields() {
808        let capture = Capture::default();
809        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
810        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
811        observer.record_domain_event(
812            call.context(),
813            DomainEvent::new("order.created")
814                .unwrap()
815                .field("order_id", "o-1")
816                .unwrap()
817                .field("item_count", 2)
818                .unwrap(),
819        );
820        call.succeed();
821        block_on(observer.flush()).unwrap();
822
823        let records = records(&capture);
824        assert_eq!(records[1]["event"], "domain.event");
825        assert_eq!(records[1]["name"], "order.created");
826        assert_eq!(records[1]["fields"]["order_id"], "o-1");
827        assert_eq!(records[1]["fields"]["item_count"], 2);
828    }
829
830    #[test]
831    fn dropped_call_is_recorded_as_abandoned() {
832        let capture = Capture::default();
833        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
834        let (call, _) = observer.start_external_call("shop", "orders", "orders", "create", None);
835        drop(call);
836        block_on(observer.flush()).unwrap();
837        let records = records(&capture);
838        assert_eq!(records[1]["outcome"], "abandoned");
839    }
840}
saddle_observability/call.rs

saddle_observability/
call.rs
