saddle_observability/
chain.rs

1use std::{error::Error, fmt, time::Instant};
2
3use saddle_core::{CallContext, ErrorKind, SaddleError};
4use serde_json::{Value, json};
5
6use crate::{EventLevel, Observer, OutputStage, logger::LogRecord};
7
8const MAX_IDENTITY_BYTES: usize = 256;
9
10#[derive(Clone, Debug, Eq, PartialEq)]
11pub enum ChainFieldError {
12    Empty,
13    TooLong,
14    ControlCharacter,
15    UnsafeAuthority,
16    ZeroAttempt,
17}
18
19impl fmt::Display for ChainFieldError {
20    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
21        f.write_str(match self {
22            Self::Empty => "observability identity must not be empty",
23            Self::TooLong => "observability identity exceeds 256 bytes",
24            Self::ControlCharacter => "observability identity contains a control character",
25            Self::UnsafeAuthority => "outbound authority must be a credential-free target label",
26            Self::ZeroAttempt => "attempt must be greater than zero",
27        })
28    }
29}
30
31impl Error for ChainFieldError {}
32
33macro_rules! safe_identity {
34    ($name:ident) => {
35        #[derive(Clone, Debug, Eq, PartialEq)]
36        pub struct $name(String);
37
38        impl $name {
39            pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
40                validate_identity(value.into()).map(Self)
41            }
42
43            pub fn as_str(&self) -> &str {
44                &self.0
45            }
46        }
47    };
48}
49
50/// Request and route identities recur in diagnostic projections. Their
51/// bounded bytes travel with the prepaid owner instead of cloning Strings.
52#[derive(Clone, Debug, Eq, PartialEq)]
53struct InlineIdentity {
54    bytes: [u8; MAX_IDENTITY_BYTES],
55    len: u16,
56}
57impl InlineIdentity {
58    fn new(value: &str) -> Result<Self, ChainFieldError> {
59        if value.is_empty() { return Err(ChainFieldError::Empty); }
60        if value.len() > MAX_IDENTITY_BYTES { return Err(ChainFieldError::TooLong); }
61        if value.chars().any(char::is_control) { return Err(ChainFieldError::ControlCharacter); }
62        let mut identity = Self { bytes: [0; MAX_IDENTITY_BYTES], len: value.len() as u16 };
63        identity.bytes[..value.len()].copy_from_slice(value.as_bytes());
64        Ok(identity)
65    }
66    fn as_str(&self) -> &str {
67        std::str::from_utf8(&self.bytes[..usize::from(self.len)]).expect("validated diagnostic identity")
68    }
69}
70macro_rules! inline_identity {
71    ($name:ident) => {
72        #[derive(Clone, Debug, Eq, PartialEq)]
73        pub struct $name(InlineIdentity);
74        impl $name {
75            pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
76                Self::from_borrowed(&value.into())
77            }
78            #[doc(hidden)]
79            pub fn from_borrowed(value: &str) -> Result<Self, ChainFieldError> {
80                InlineIdentity::new(value).map(Self)
81            }
82            pub fn as_str(&self) -> &str { self.0.as_str() }
83        }
84    };
85}
86inline_identity!(RequestIdentity);
87inline_identity!(RouteIdentity);
88safe_identity!(BottleneckIdentity);
89safe_identity!(RejectReason);
90
91#[derive(Clone, Debug, Eq, PartialEq)]
92pub struct OutboundAuthority(String);
93
94impl OutboundAuthority {
95    pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
96        let value = validate_identity(value.into())?;
97        if value.contains(['@', '?', '#', '/', '\\']) || value.contains("://") {
98            return Err(ChainFieldError::UnsafeAuthority);
99        }
100        Ok(Self(value))
101    }
102}
103
104fn validate_identity(value: String) -> Result<String, ChainFieldError> {
105    if value.is_empty() {
106        return Err(ChainFieldError::Empty);
107    }
108    if value.len() > MAX_IDENTITY_BYTES {
109        return Err(ChainFieldError::TooLong);
110    }
111    if value.chars().any(char::is_control) {
112        return Err(ChainFieldError::ControlCharacter);
113    }
114    Ok(value)
115}
116
117#[derive(Clone, Copy, Debug, Eq, PartialEq)]
118pub enum Stage {
119    Ingress,
120    Admission,
121    Handler,
122    Database,
123    ProfuseContract,
124    Response,
125    ResourceFinalization,
126}
127
128impl Stage {
129    const fn as_str(self) -> &'static str {
130        match self {
131            Self::Ingress => "ingress",
132            Self::Admission => "admission",
133            Self::Handler => "handler",
134            Self::Database => "database",
135            Self::ProfuseContract => "profuse_contract",
136            Self::Response => "response",
137            Self::ResourceFinalization => "resource_finalization",
138        }
139    }
140}
141
142#[derive(Clone, Copy, Debug, Eq, PartialEq)]
143pub enum StageOutcome {
144    Success,
145    Failure,
146    Rejected,
147    Cancelled,
148}
149
150/// DB owning-driver facts, not an inference from HTTP or connection disposal.
151#[derive(Clone, Copy, Debug, Eq, PartialEq)]
152pub enum TransactionOutcome {
153    /// COMMIT acknowledgement was received, even if the client later disconnects.
154    Committed,
155    /// No possible commit was entered. This does not prove rollback acknowledgement.
156    Rejected,
157    /// COMMIT may have executed but its acknowledgement is unavailable.
158    Unknown,
159}
160
161impl TransactionOutcome {
162    const fn as_str(self) -> &'static str {
163        match self {
164            Self::Committed => "Committed",
165            Self::Rejected => "Rejected",
166            Self::Unknown => "Unknown",
167        }
168    }
169}
170
171/// Same-request/scope carrier checked by Core and transported by Runtime.
172/// It is consumed once; it is not proof of transaction outcome or log durability.
173pub type TransactionTerminalObservation =
174    saddle_core::DbScopeTerminalObservation<(Observer, CallContext, EventContext)>;
175
176impl StageOutcome {
177    const fn as_str(self) -> &'static str {
178        match self {
179            Self::Success => "success",
180            Self::Failure => "failure",
181            Self::Rejected => "rejected",
182            Self::Cancelled => "cancelled",
183        }
184    }
185}
186
187#[derive(Clone, Debug, Eq, PartialEq)]
188pub struct EventContext {
189    request: RequestIdentity,
190    route: RouteIdentity,
191    attempt: u32,
192}
193
194impl EventContext {
195    pub(crate) fn diagnostic_attempt(&self) -> u32 {
196        self.attempt
197    }
198    pub(crate) fn diagnostic_request(&self) -> &str {
199        self.request.as_str()
200    }
201    pub(crate) fn diagnostic_route(&self) -> &str {
202        self.route.as_str()
203    }
204    pub fn new(
205        request: RequestIdentity,
206        route: RouteIdentity,
207        attempt: u32,
208    ) -> Result<Self, ChainFieldError> {
209        if attempt == 0 {
210            return Err(ChainFieldError::ZeroAttempt);
211        }
212        Ok(Self {
213            request,
214            route,
215            attempt,
216        })
217    }
218}
219
220pub struct ActiveStage {
221    observer: Observer,
222    context: CallContext,
223    parent_span: String,
224    stage: Stage,
225    request: RequestIdentity,
226    route: RouteIdentity,
227    attempt: u32,
228    started_at: Instant,
229    finished: bool,
230}
231
232impl Observer {
233    /// DB owning driver calls once before returning its result/physical receipt.
234    /// Uses the carrier's observer/context, with no caller-supplied scope or sink.
235    /// Queue loss preserves the DB result; a missing log remains NOT_PROVEN.
236    ///
237    /// ```compile_fail
238    /// use saddle_observability::{Observer, TransactionOutcome, TransactionTerminalObservation};
239    /// fn replay(observation: TransactionTerminalObservation) {
240    ///     Observer::record_transaction_terminal(observation, TransactionOutcome::Committed);
241    ///     Observer::record_transaction_terminal(observation, TransactionOutcome::Rejected);
242    /// }
243    /// ```
244    pub fn record_transaction_terminal(
245        observation: TransactionTerminalObservation,
246        outcome: TransactionOutcome,
247    ) {
248        let ((observer, context, event_context), scope) = observation.into_log_parts();
249        let mut record = base_record(
250            &context,
251            if outcome == TransactionOutcome::Committed {
252                EventLevel::Info
253            } else {
254                EventLevel::Error
255            },
256            "database_transaction_terminal",
257            "database",
258        );
259        add_event_context(&mut record, &event_context);
260        // Core's concrete Serialize-only payload has exactly one numeric field.
261        // A serialization failure drops this attempt; it never changes DB outcome.
262        let Ok(Value::Object(fields)) = serde_json::to_value(scope) else {
263            return;
264        };
265        record.data.extend(fields);
266        record.data.insert(
267            "transaction_outcome".into(),
268            Value::String(outcome.as_str().into()),
269        );
270        record
271            .data
272            .insert("outcome".into(), Value::String(outcome.as_str().into()));
273        observer.emit(record);
274    }
275
276    pub fn record_lifecycle_timeout(
277        &self,
278        application: &str,
279        stage: LifecycleTimeoutStage,
280        elapsed_ms: u64,
281    ) {
282        let Ok((call, _)) = self.start_external_call_checked(
283            application,
284            "runtime",
285            "lifecycle",
286            stage.as_str(),
287            None,
288        ) else {
289            return;
290        };
291        let mut record = base_record(
292            call.context(),
293            EventLevel::Error,
294            "framework.lifecycle.timeout",
295            "lifecycle",
296        );
297        record
298            .data
299            .insert("timeout_stage".into(), Value::String(stage.as_str().into()));
300        record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
301        record
302            .data
303            .insert("outcome".into(), Value::String("timeout".into()));
304        self.emit(record);
305        call.fail(&SaddleError::new(
306            ErrorKind::Infrastructure,
307            "runtime.lifecycle_timeout",
308            "managed lifecycle deadline elapsed",
309        ));
310    }
311
312    pub fn start_stage(
313        &self,
314        parent: &CallContext,
315        stage: Stage,
316        event_context: EventContext,
317    ) -> ActiveStage {
318        let context = CallContext::new(
319            parent.application().clone(),
320            parent.module().clone(),
321            parent.service().clone(),
322            parent.operation().clone(),
323            parent.trace_id(),
324            self.new_span_id(),
325        )
326        .with_trace_correlation_id(parent.trace_correlation_id().clone())
327        .with_rpc_correlation_id(parent.rpc_correlation_id().cloned());
328        let active = ActiveStage {
329            observer: self.clone(),
330            context,
331            parent_span: parent.span_id().to_string(),
332            stage,
333            request: event_context.request,
334            route: event_context.route,
335            attempt: event_context.attempt,
336            started_at: Instant::now(),
337            finished: false,
338        };
339        active.emit_started();
340        active
341    }
342
343    pub fn record_capacity(
344        &self,
345        context: &CallContext,
346        event_context: &EventContext,
347        value: CapacityObservation,
348    ) {
349        self.inner
350            .metrics
351            .capacity(value.dimension, value.used, value.reject_reason.is_some());
352        let mut record = base_record(context, EventLevel::Info, "framework.capacity", "admission");
353        add_event_context(&mut record, event_context);
354        if let Some(dimension) = value.dimension {
355            record.data.insert(
356                "capacity_dimension".into(),
357                Value::String(dimension.as_str().into()),
358            );
359        }
360        record.data.insert("budget".into(), json!(value.budget));
361        record.data.insert("limit".into(), json!(value.limit));
362        record.data.insert("used".into(), json!(value.used));
363        record
364            .data
365            .insert("elapsed_ms".into(), json!(value.elapsed_ms));
366        record
367            .data
368            .insert("bottleneck".into(), Value::String(value.bottleneck.0));
369        record.data.insert(
370            "outcome".into(),
371            Value::String(
372                if value.reject_reason.is_some() {
373                    "rejected"
374                } else {
375                    "accepted"
376                }
377                .into(),
378            ),
379        );
380        if let Some(reason) = value.reject_reason {
381            record
382                .data
383                .insert("reject_reason".into(), Value::String(reason.0));
384        }
385        self.emit(record);
386    }
387
388    /// Update the existing fixed-label counter without allocating or emitting a
389    /// second log. The DB owning terminal path calls this exactly once with its
390    /// actual physical disposition, independently of operation outcome.
391    pub fn observe_database_disposition(&self, disposition: DatabaseDisposition) {
392        self.inner.metrics.database(disposition);
393    }
394
395    pub fn record_database_disposition(
396        &self,
397        context: &CallContext,
398        event_context: &EventContext,
399        disposition: DatabaseDisposition,
400    ) {
401        self.observe_database_disposition(disposition);
402        let mut record = base_record(
403            context,
404            EventLevel::Info,
405            "framework.database.disposition",
406            "database",
407        );
408        add_event_context(&mut record, event_context);
409        record.data.insert(
410            "db_disposition".into(),
411            Value::String(disposition.as_str().into()),
412        );
413        record
414            .data
415            .insert("outcome".into(), Value::String(disposition.as_str().into()));
416        self.emit(record);
417    }
418
419    pub fn record_resource_finalization(
420        &self,
421        context: &CallContext,
422        event_context: &EventContext,
423        disposition: DatabaseDisposition,
424        elapsed_ms: u64,
425    ) {
426        let mut record = base_record(
427            context,
428            EventLevel::Info,
429            "framework.resource.finalized",
430            "resource_finalization",
431        );
432        add_event_context(&mut record, event_context);
433        record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
434        record
435            .data
436            .insert("credit".into(), Value::String("released".into()));
437        record.data.insert(
438            "db_disposition".into(),
439            Value::String(disposition.as_str().into()),
440        );
441        record
442            .data
443            .insert("outcome".into(), Value::String("success".into()));
444        self.emit(record);
445    }
446
447    pub fn record_outbound(
448        &self,
449        context: &CallContext,
450        event_context: &EventContext,
451        observation: OutboundObservation,
452    ) {
453        self.inner.metrics.outbound(observation.result);
454        let mut record = base_record(
455            context,
456            EventLevel::Info,
457            "framework.outbound",
458            "profuse_contract",
459        );
460        add_event_context(&mut record, event_context);
461        record
462            .data
463            .insert("zone".into(), Value::String(observation.zone.as_str().to_owned()));
464        record
465            .data
466            .insert("authority".into(), Value::String(observation.authority.0));
467        record.data.insert(
468            "outbound_result".into(),
469            Value::String(observation.result.as_str().into()),
470        );
471        record.data.insert(
472            "outcome".into(),
473            Value::String(observation.result.as_str().into()),
474        );
475        self.emit(record);
476    }
477
478    pub fn record_lifecycle(
479        &self,
480        context: &CallContext,
481        event_context: &EventContext,
482        state: LifecycleState,
483        health: Health,
484    ) {
485        self.inner.metrics.lifecycle(state, health);
486        let mut record = base_record(
487            context,
488            if health == Health::Healthy {
489                EventLevel::Info
490            } else {
491                EventLevel::Error
492            },
493            "framework.lifecycle",
494            "resource_finalization",
495        );
496        add_event_context(&mut record, event_context);
497        record
498            .data
499            .insert("lifecycle".into(), Value::String(state.as_str().into()));
500        record
501            .data
502            .insert("health".into(), Value::String(health.as_str().into()));
503        record
504            .data
505            .insert("outcome".into(), Value::String(health.as_str().into()));
506        self.emit(record);
507    }
508
509    pub fn record_logger_health(
510        &self,
511        context: &CallContext,
512        event_context: &EventContext,
513        dropped: u64,
514        failure: Option<OutputStage>,
515    ) {
516        self.inner.metrics.logger_output(failure.is_some());
517        let mut record = base_record(
518            context,
519            if failure.is_some() || dropped > 0 {
520                EventLevel::Error
521            } else {
522                EventLevel::Info
523            },
524            "framework.logger.health",
525            "logger",
526        );
527        add_event_context(&mut record, event_context);
528        record.data.insert("logger_dropped".into(), json!(dropped));
529        record.data.insert(
530            "logger_health".into(),
531            Value::String(
532                if failure.is_some() {
533                    "output_failed"
534                } else {
535                    "healthy"
536                }
537                .into(),
538            ),
539        );
540        record.data.insert(
541            "outcome".into(),
542            Value::String(
543                if failure.is_some() {
544                    "failure"
545                } else {
546                    "success"
547                }
548                .into(),
549            ),
550        );
551        if let Some(stage) = failure {
552            record.data.insert(
553                "output_failed".into(),
554                Value::String(format!("{stage:?}").to_ascii_lowercase()),
555            );
556        }
557        self.emit(record);
558    }
559}
560
561#[derive(Clone, Copy, Debug, Eq, PartialEq)]
562pub enum LifecycleTimeoutStage {
563    ComponentStart,
564    RequestDrain,
565    ComponentShutdown,
566    PostDriverFinalization,
567}
568
569impl LifecycleTimeoutStage {
570    const fn as_str(self) -> &'static str {
571        match self {
572            Self::ComponentStart => "component_start",
573            Self::RequestDrain => "request_drain",
574            Self::ComponentShutdown => "component_shutdown",
575            Self::PostDriverFinalization => "post_driver_finalization",
576        }
577    }
578}
579
580impl ActiveStage {
581    /// Finish a non-technical result without inventing a source receipt. Only
582    /// Succeeded/Rejected are accepted; all other axes return the untouched stage
583    /// for the caller to finish through its actual diagnostic receipt path.
584    #[allow(clippy::result_large_err)] // Returning the owner must not allocate.
585    pub fn finish_nonfailure_bounded(
586        mut self,
587        output: Option<&crate::EmergencyDiagnosticHandle>,
588        axes: &saddle_core::DiagnosticOutcomeAxes,
589    ) -> Result<crate::DiagnosticSubmission, Self> {
590        if !matches!(
591            axes.operation,
592            saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
593        ) {
594            return Err(self);
595        }
596        let elapsed = self.finish_bounded_metrics(axes);
597        Ok(
598            output.map_or(crate::DiagnosticSubmission::OutputUnavailable, |output| {
599                output.submit_boundary_parts(
600                    None,
601                    axes,
602                    Some((&self.context, self.request.as_str(), self.route.as_str())),
603                    Some((self.stage.as_str(), elapsed)),
604                )
605            }),
606        )
607    }
608
609    /// Ends this stage once using the source receipt's original context. Missing
610    /// output changes delivery status only, never the operation or Drop outcome.
611    pub fn finish_request_failure<D>(
612        mut self,
613        output: Option<&crate::EmergencyDiagnosticHandle>,
614        axes: &saddle_core::DiagnosticOutcomeAxes,
615        reference: &crate::RequestBoundaryReference<D>,
616    ) -> crate::BoundaryDiagnosticDelivery {
617        let elapsed = self.finish_bounded_metrics(axes);
618        reference.record_stage(output, axes, self.stage.as_str(), elapsed)
619    }
620
621    fn finish_bounded_metrics(&mut self, axes: &saddle_core::DiagnosticOutcomeAxes) -> u64 {
622        self.finished = true;
623        let elapsed = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
624        use saddle_core::OperationOutcome;
625        let metric = match axes.operation {
626            OperationOutcome::Succeeded => Some(StageOutcome::Success),
627            OperationOutcome::Rejected => Some(StageOutcome::Rejected),
628            OperationOutcome::Failed | OperationOutcome::Panicked => Some(StageOutcome::Failure),
629            OperationOutcome::Cancelled | OperationOutcome::TimedOut => {
630                Some(StageOutcome::Cancelled)
631            }
632            OperationOutcome::Unknown => None,
633        };
634        if let Some(metric) = metric {
635            self.observer
636                .inner
637                .metrics
638                .stage_finished(self.stage, metric, elapsed);
639        }
640        elapsed
641    }
642    /// Finish from independent source-owned axes without constructing SaddleError
643    /// or allocating a LogRecord. Delivery failure never retries the stage.
644    pub fn finish_bounded(
645        self,
646        output: &crate::EmergencyDiagnosticHandle,
647        axes: &saddle_core::DiagnosticOutcomeAxes,
648        diagnostic: Option<&saddle_core::BoundedDiagnostic>,
649    ) -> crate::DiagnosticSubmission {
650        self.finish_bounded_reference(output, axes, diagnostic.map(|d| d.occurrence()))
651    }
652    /// Finish with correlation only; never repeats the source detail.
653    pub fn finish_bounded_reference(
654        mut self,
655        output: &crate::EmergencyDiagnosticHandle,
656        axes: &saddle_core::DiagnosticOutcomeAxes,
657        occurrence: Option<saddle_core::DiagnosticOccurrence>,
658    ) -> crate::DiagnosticSubmission {
659        let elapsed = self.finish_bounded_metrics(axes);
660        output.submit_boundary_parts(
661            occurrence,
662            axes,
663            Some((&self.context, self.request.as_str(), self.route.as_str())),
664            Some((self.stage.as_str(), elapsed)),
665        )
666    }
667    pub fn context(&self) -> &CallContext {
668        &self.context
669    }
670    pub fn succeed(mut self) {
671        self.finish(StageOutcome::Success, None);
672    }
673    pub fn reject(mut self, error: &SaddleError) {
674        self.finish(StageOutcome::Rejected, Some(error));
675    }
676    pub fn fail(mut self, error: &SaddleError) {
677        self.finish(StageOutcome::Failure, Some(error));
678    }
679
680    fn emit_started(&self) {
681        let mut record = self.record(EventLevel::Info, "framework.stage.started");
682        record
683            .data
684            .insert("outcome".into(), Value::String("started".into()));
685        self.observer.emit(record);
686    }
687
688    fn finish(&mut self, outcome: StageOutcome, error: Option<&SaddleError>) {
689        if self.finished {
690            return;
691        }
692        let mut record = self.record(
693            if outcome == StageOutcome::Success {
694                EventLevel::Info
695            } else {
696                EventLevel::Error
697            },
698            "framework.stage.finished",
699        );
700        record
701            .data
702            .insert("outcome".into(), Value::String(outcome.as_str().into()));
703        let elapsed_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
704        record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
705        if let Some(error) = error {
706            record
707                .data
708                .insert("error_code".into(), Value::String(error.code().to_owned()));
709            record.data.insert(
710                "error_kind".into(),
711                Value::String(error_kind(error.kind()).into()),
712            );
713        }
714        self.observer
715            .inner
716            .metrics
717            .stage_finished(self.stage, outcome, elapsed_ms);
718        self.observer.emit(record);
719        self.finished = true;
720    }
721
722    fn record(&self, level: EventLevel, event: &'static str) -> LogRecord {
723        let mut record = base_record(&self.context, level, event, self.stage.as_str());
724        record.parent = Some(self.parent_span.clone());
725        record.parent_span_id = Some(self.parent_span.clone());
726        record.data.insert(
727            "request_identity".into(),
728            Value::String(self.request.as_str().to_owned()),
729        );
730        record
731            .data
732            .insert("route".into(), Value::String(self.route.as_str().to_owned()));
733        record.data.insert("attempt".into(), json!(self.attempt));
734        record.data.insert("elapsed_ms".into(), json!(0));
735        record
736            .data
737            .insert("error_code".into(), Value::String("none".into()));
738        record
739    }
740}
741
742impl Drop for ActiveStage {
743    fn drop(&mut self) {
744        self.finish(StageOutcome::Cancelled, None);
745    }
746}
747
748pub struct CapacityObservation {
749    dimension: Option<CapacityDimension>,
750    budget: u64,
751    limit: u64,
752    used: u64,
753    bottleneck: BottleneckIdentity,
754    reject_reason: Option<RejectReason>,
755    elapsed_ms: u64,
756}
757
758impl CapacityObservation {
759    pub fn accepted(budget: u64, limit: u64, used: u64, bottleneck: BottleneckIdentity) -> Self {
760        Self {
761            dimension: None,
762            budget,
763            limit,
764            used,
765            bottleneck,
766            reject_reason: None,
767            elapsed_ms: 0,
768        }
769    }
770    pub fn rejected(
771        budget: u64,
772        limit: u64,
773        used: u64,
774        bottleneck: BottleneckIdentity,
775        reason: RejectReason,
776    ) -> Self {
777        Self {
778            dimension: None,
779            budget,
780            limit,
781            used,
782            bottleneck,
783            reject_reason: Some(reason),
784            elapsed_ms: 0,
785        }
786    }
787
788    pub fn accepted_dimension(
789        dimension: CapacityDimension,
790        budget: u64,
791        limit: u64,
792        used: u64,
793        bottleneck: BottleneckIdentity,
794        elapsed_ms: u64,
795    ) -> Self {
796        Self {
797            dimension: Some(dimension),
798            budget,
799            limit,
800            used,
801            bottleneck,
802            reject_reason: None,
803            elapsed_ms,
804        }
805    }
806
807    pub fn rejected_dimension(
808        dimension: CapacityDimension,
809        budget: u64,
810        limit: u64,
811        used: u64,
812        bottleneck: BottleneckIdentity,
813        reason: RejectReason,
814        elapsed_ms: u64,
815    ) -> Self {
816        Self {
817            dimension: Some(dimension),
818            budget,
819            limit,
820            used,
821            bottleneck,
822            reject_reason: Some(reason),
823            elapsed_ms,
824        }
825    }
826}
827
828#[derive(Clone, Copy, Debug, Eq, PartialEq)]
829pub enum CapacityDimension {
830    Cpu,
831    Memory,
832    Database,
833    ProfuseContract,
834}
835
836impl CapacityDimension {
837    const fn as_str(self) -> &'static str {
838        match self {
839            Self::Cpu => "cpu",
840            Self::Memory => "memory",
841            Self::Database => "database",
842            Self::ProfuseContract => "profuse_contract",
843        }
844    }
845}
846
847#[derive(Clone, Copy, Debug, Eq, PartialEq)]
848pub enum DatabaseDisposition {
849    NotUsed,
850    Returned,
851    Discarded,
852}
853impl DatabaseDisposition {
854    const fn as_str(self) -> &'static str {
855        match self {
856            Self::NotUsed => "not_used",
857            Self::Returned => "returned",
858            Self::Discarded => "discarded",
859        }
860    }
861}
862
863pub struct OutboundObservation {
864    zone: RouteIdentity,
865    authority: OutboundAuthority,
866    result: OutboundResult,
867}
868impl OutboundObservation {
869    pub fn new(zone: RouteIdentity, authority: OutboundAuthority, result: OutboundResult) -> Self {
870        Self {
871            zone,
872            authority,
873            result,
874        }
875    }
876}
877
878#[derive(Clone, Copy, Debug, Eq, PartialEq)]
879pub enum OutboundResult {
880    Success,
881    Failure,
882    Rejected,
883    Timeout,
884}
885impl OutboundResult {
886    const fn as_str(self) -> &'static str {
887        match self {
888            Self::Success => "success",
889            Self::Failure => "failure",
890            Self::Rejected => "rejected",
891            Self::Timeout => "timeout",
892        }
893    }
894}
895
896#[derive(Clone, Copy, Debug, Eq, PartialEq)]
897pub enum LifecycleState {
898    Starting,
899    Running,
900    Draining,
901    Stopped,
902}
903impl LifecycleState {
904    const fn as_str(self) -> &'static str {
905        match self {
906            Self::Starting => "starting",
907            Self::Running => "running",
908            Self::Draining => "draining",
909            Self::Stopped => "stopped",
910        }
911    }
912}
913
914#[derive(Clone, Copy, Debug, Eq, PartialEq)]
915pub enum Health {
916    Healthy,
917    Degraded,
918    Failed,
919}
920impl Health {
921    const fn as_str(self) -> &'static str {
922        match self {
923            Self::Healthy => "healthy",
924            Self::Degraded => "degraded",
925            Self::Failed => "failed",
926        }
927    }
928}
929
930fn base_record(
931    context: &CallContext,
932    level: EventLevel,
933    event: &'static str,
934    stage: &'static str,
935) -> LogRecord {
936    let mut record = LogRecord::new(level, event);
937    record.trace_id = Some(context.trace_correlation_id().to_string());
938    record.span = Some(stage.into());
939    record.span_id = Some(context.span_id().to_string());
940    record
941        .data
942        .insert("timestamp".into(), json!(record.timestamp_unix_ms));
943    record
944        .data
945        .insert("stage".into(), Value::String(stage.into()));
946    record.data.insert(
947        "rpc_id".into(),
948        context
949            .rpc_correlation_id()
950            .map(|rpc| Value::String(rpc.as_str().into()))
951            .unwrap_or(Value::Null),
952    );
953    record.data.insert(
954        "request".into(),
955        Value::String(context.operation().to_string()),
956    );
957    record
958}
959
960fn add_event_context(record: &mut LogRecord, context: &EventContext) {
961    record.data.insert(
962        "request_identity".into(),
963        Value::String(context.request.as_str().to_owned()),
964    );
965    record
966        .data
967        .insert("route".into(), Value::String(context.route.as_str().to_owned()));
968    record.data.insert("attempt".into(), json!(context.attempt));
969    record.data.insert("elapsed_ms".into(), json!(0));
970    record
971        .data
972        .insert("error_code".into(), Value::String("none".into()));
973}
974
975const fn error_kind(kind: ErrorKind) -> &'static str {
976    match kind {
977        ErrorKind::InvalidArgument => "invalid_argument",
978        ErrorKind::NotFound => "not_found",
979        ErrorKind::Conflict => "conflict",
980        ErrorKind::Business => "business",
981        ErrorKind::Unavailable => "unavailable",
982        ErrorKind::Infrastructure => "infrastructure",
983        ErrorKind::Internal => "internal",
984        _ => "unknown",
985    }
986}
987
988#[cfg(test)]
989mod tests {
990    use std::{
991        future::Future,
992        io,
993        sync::{Arc, Mutex},
994        task::{Context, Poll, Wake, Waker},
995        thread,
996    };
997
998    use super::*;
999    use crate::ObserverConfig;
1000
1001    #[derive(Clone, Default)]
1002    struct Capture(Arc<Mutex<Vec<u8>>>);
1003
1004    impl io::Write for Capture {
1005        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
1006            self.0.lock().unwrap().extend_from_slice(bytes);
1007            Ok(bytes.len())
1008        }
1009        fn flush(&mut self) -> io::Result<()> {
1010            Ok(())
1011        }
1012    }
1013
1014    struct ThreadWaker(thread::Thread);
1015    impl Wake for ThreadWaker {
1016        fn wake(self: Arc<Self>) {
1017            self.0.unpark();
1018        }
1019    }
1020
1021    fn block_on<T>(future: impl Future<Output = T>) -> T {
1022        let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
1023        let mut context = Context::from_waker(&waker);
1024        let mut future = std::pin::pin!(future);
1025        loop {
1026            match future.as_mut().poll(&mut context) {
1027                Poll::Ready(output) => return output,
1028                Poll::Pending => thread::park(),
1029            }
1030        }
1031    }
1032
1033    fn event_context(attempt: u32) -> EventContext {
1034        EventContext::new(
1035            RequestIdentity::new("request-7").unwrap(),
1036            RouteIdentity::new("orders.create").unwrap(),
1037            attempt,
1038        )
1039        .unwrap()
1040    }
1041
1042    #[test]
1043    fn transaction_terminal_serial_scope_schema_and_foreign_restore() {
1044        let capture = Capture::default();
1045        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
1046        let (root, _) = observer
1047            .start_external_call_with_rpc(
1048                "shop",
1049                "entry",
1050                "orders",
1051                "create",
1052                Some("safe-trace"),
1053                saddle_core::RpcCorrelationId::new("0").unwrap(),
1054            )
1055            .unwrap();
1056        let (startup, issuer) =
1057            saddle_core::DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
1058        let process = startup.into_process_capability();
1059        let (mut request, mut execution) = issuer.issue_request().unwrap();
1060        let (_, foreign) = issuer.issue_request().unwrap();
1061        for outcome in [
1062            TransactionOutcome::Committed,
1063            TransactionOutcome::Rejected,
1064            TransactionOutcome::Unknown,
1065        ] {
1066            let observation = request
1067                .take_scope_observation(
1068                    &execution,
1069                    (observer.clone(), root.context().clone(), event_context(1)),
1070                )
1071                .ok()
1072                .unwrap();
1073            let observation = observation.bind_terminal(&foreign).err().unwrap();
1074            let observation = observation.bind_terminal(&execution).ok().unwrap();
1075            Observer::record_transaction_terminal(observation, outcome);
1076            assert!(request.take_scope_observation(&execution, ()).is_err());
1077            let physical = process.connection_returned(execution, ()).ok().unwrap();
1078            let (next, _, _) = saddle_core::pair_db_physical_disposition(physical, request)
1079                .ok()
1080                .unwrap()
1081                .into_scope_continuation();
1082            (request, execution) = next.into_next_scope().ok().unwrap();
1083        }
1084        block_on(observer.flush()).unwrap();
1085        let rows: Vec<_> = records(&capture)
1086            .into_iter()
1087            .filter(|r| r["event"] == "database_transaction_terminal")
1088            .collect();
1089        assert_eq!(rows.len(), 3);
1090        for (index, outcome) in ["Committed", "Rejected", "Unknown"].into_iter().enumerate() {
1091            let row = &rows[index];
1092            assert_eq!(row["transaction_scope"], index);
1093            assert_eq!(row["transaction_outcome"], outcome);
1094            assert_eq!(row["trace_id"], "safe-trace");
1095            assert_eq!(row["rpc_id"], "0");
1096            assert_eq!(row["span_id"], root.context().span_id().to_string());
1097            assert_ne!(row["rpc_id"], row["span_id"]);
1098            assert_eq!(row["request_identity"], "request-7");
1099            assert_eq!(row["route"], "orders.create");
1100            let mut keys: Vec<_> = row
1101                .as_object()
1102                .unwrap()
1103                .keys()
1104                .map(String::as_str)
1105                .collect();
1106            keys.sort_unstable();
1107            let mut expected = vec![
1108                "timestamp_unix_ms",
1109                "timestamp",
1110                "level",
1111                "event",
1112                "stage",
1113                "trace_id",
1114                "rpc_id",
1115                "span",
1116                "span_id",
1117                "request",
1118                "request_identity",
1119                "route",
1120                "attempt",
1121                "elapsed_ms",
1122                "error_code",
1123                "outcome",
1124                "transaction_scope",
1125                "transaction_outcome",
1126            ];
1127            expected.sort_unstable();
1128            assert_eq!(keys, expected); // No SQL/parameters/business error or arbitrary fields.
1129        }
1130        root.succeed();
1131        block_on(observer.shutdown()).unwrap();
1132    }
1133
1134    #[test]
1135    fn transaction_terminal_full_queue_never_waits_for_writer() {
1136        use std::sync::mpsc;
1137        use std::time::Duration;
1138        struct PausedWriter {
1139            entered: Option<mpsc::Sender<()>>,
1140            release: mpsc::Receiver<()>,
1141        }
1142        impl io::Write for PausedWriter {
1143            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
1144                if let Some(entered) = self.entered.take() {
1145                    entered.send(()).unwrap();
1146                    self.release.recv().unwrap();
1147                }
1148                Ok(bytes.len())
1149            }
1150            fn flush(&mut self) -> io::Result<()> {
1151                Ok(())
1152            }
1153        }
1154        let (entered, reached) = mpsc::channel();
1155        let (release, gate) = mpsc::channel();
1156        let observer = Observer::with_writer(
1157            ObserverConfig { queue_capacity: 1 },
1158            PausedWriter {
1159                entered: Some(entered),
1160                release: gate,
1161            },
1162        )
1163        .unwrap();
1164        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
1165        reached.recv_timeout(Duration::from_secs(5)).unwrap();
1166        observer.emit(LogRecord::new(EventLevel::Info, "fill"));
1167        let (_, issuer) =
1168            saddle_core::DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
1169        let (mut request, execution) = issuer.issue_request().unwrap();
1170        let observation = request
1171            .take_scope_observation(
1172                &execution,
1173                (observer.clone(), root.context().clone(), event_context(1)),
1174            )
1175            .ok()
1176            .unwrap()
1177            .bind_terminal(&execution)
1178            .ok()
1179            .unwrap();
1180        let (done, completion) = mpsc::channel();
1181        let emitter = thread::spawn(move || {
1182            Observer::record_transaction_terminal(observation, TransactionOutcome::Unknown);
1183            done.send(()).unwrap();
1184        });
1185        let result = completion.recv_timeout(Duration::from_secs(2));
1186        let dropped = observer.dropped_events();
1187        release.send(()).unwrap(); // Release even if a regression blocked the emitter.
1188        emitter.join().unwrap();
1189        assert!(result.is_ok());
1190        assert_eq!(dropped, 1);
1191        let _ = block_on(observer.flush());
1192        root.succeed();
1193        let _ = block_on(observer.shutdown());
1194    }
1195
1196    fn records(capture: &Capture) -> Vec<Value> {
1197        String::from_utf8(capture.0.lock().unwrap().clone())
1198            .unwrap()
1199            .lines()
1200            .map(|line| serde_json::from_str(line).unwrap())
1201            .collect()
1202    }
1203
1204    #[test]
1205    fn lifecycle_timeout_is_typed_and_contains_no_application_payload() {
1206        let capture = Capture::default();
1207        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
1208        observer.record_lifecycle_timeout(
1209            "profusegw",
1210            LifecycleTimeoutStage::ComponentShutdown,
1211            30_000,
1212        );
1213        block_on(observer.flush()).unwrap();
1214
1215        let records = records(&capture);
1216        let timeout = records
1217            .iter()
1218            .find(|record| record["event"] == "framework.lifecycle.timeout")
1219            .unwrap();
1220        assert_eq!(timeout["timeout_stage"], "component_shutdown");
1221        assert_eq!(timeout["elapsed_ms"], 30_000);
1222        assert_eq!(timeout["outcome"], "timeout");
1223        assert!(timeout.get("payload").is_none());
1224    }
1225
1226    #[test]
1227    fn stage_chain_is_correlated_ordered_and_has_one_terminal() {
1228        let capture = Capture::default();
1229        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
1230        let (root, _) = observer.start_external_call(
1231            "shop",
1232            "entry",
1233            "orders",
1234            "create",
1235            Some("00112233445566778899aabbccddeeff"),
1236        );
1237        let stages = [
1238            Stage::Ingress,
1239            Stage::Admission,
1240            Stage::Handler,
1241            Stage::Database,
1242            Stage::ProfuseContract,
1243            Stage::Response,
1244            Stage::ResourceFinalization,
1245        ];
1246        for (index, stage) in stages.into_iter().enumerate() {
1247            observer
1248                .start_stage(root.context(), stage, event_context((index + 1) as u32))
1249                .succeed();
1250        }
1251        drop(observer.start_stage(root.context(), Stage::Handler, event_context(8)));
1252        root.succeed();
1253        block_on(observer.flush()).unwrap();
1254
1255        let records = records(&capture);
1256        let stage_records: Vec<_> = records
1257            .iter()
1258            .filter(|value| {
1259                value["event"]
1260                    .as_str()
1261                    .unwrap()
1262                    .starts_with("framework.stage.")
1263            })
1264            .collect();
1265        assert_eq!(stage_records.len(), 16);
1266        for pair in stage_records.chunks_exact(2) {
1267            assert_eq!(pair[0]["event"], "framework.stage.started");
1268            assert_eq!(pair[1]["event"], "framework.stage.finished");
1269            assert_eq!(pair[0]["rpc_id"], pair[1]["rpc_id"]);
1270            assert_eq!(pair[0]["trace_id"], "00112233445566778899aabbccddeeff");
1271            for field in [
1272                "timestamp_unix_ms",
1273                "timestamp",
1274                "level",
1275                "event",
1276                "stage",
1277                "trace_id",
1278                "rpc_id",
1279                "request_identity",
1280                "route",
1281                "attempt",
1282                "outcome",
1283                "error_code",
1284            ] {
1285                assert!(pair[0].get(field).is_some(), "missing {field}");
1286            }
1287            assert!(pair[1].get("elapsed_ms").is_some());
1288        }
1289        assert_eq!(stage_records.last().unwrap()["outcome"], "cancelled");
1290    }
1291
1292    #[test]
1293    fn closed_observations_have_common_fields_and_no_sensitive_payload() {
1294        let capture = Capture::default();
1295        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
1296        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
1297        let common = event_context(1);
1298        observer.record_capacity(
1299            root.context(),
1300            &common,
1301            CapacityObservation::rejected(
1302                100,
1303                80,
1304                80,
1305                BottleneckIdentity::new("request_slots").unwrap(),
1306                RejectReason::new("limit_reached").unwrap(),
1307            ),
1308        );
1309        observer.record_database_disposition(
1310            root.context(),
1311            &common,
1312            DatabaseDisposition::Returned,
1313        );
1314        observer.record_outbound(
1315            root.context(),
1316            &common,
1317            OutboundObservation::new(
1318                RouteIdentity::new("cn-hz-a").unwrap(),
1319                OutboundAuthority::new("inventory-service").unwrap(),
1320                OutboundResult::Success,
1321            ),
1322        );
1323        observer.record_lifecycle(
1324            root.context(),
1325            &common,
1326            LifecycleState::Draining,
1327            Health::Healthy,
1328        );
1329        observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
1330        root.succeed();
1331        let _ = block_on(observer.flush());
1332
1333        let records: Vec<_> = records(&capture)
1334            .into_iter()
1335            .filter(|value| {
1336                matches!(
1337                    value["event"].as_str(),
1338                    Some(
1339                        "framework.capacity"
1340                            | "framework.database.disposition"
1341                            | "framework.outbound"
1342                            | "framework.lifecycle"
1343                            | "framework.logger.health"
1344                    )
1345                )
1346            })
1347            .collect();
1348        assert_eq!(records.len(), 5);
1349        for record in records {
1350            for field in [
1351                "timestamp_unix_ms",
1352                "timestamp",
1353                "level",
1354                "event",
1355                "stage",
1356                "trace_id",
1357                "rpc_id",
1358                "request_identity",
1359                "route",
1360                "attempt",
1361                "elapsed_ms",
1362                "outcome",
1363                "error_code",
1364            ] {
1365                assert!(record.get(field).is_some(), "missing {field}");
1366            }
1367            let encoded = serde_json::to_string(&record).unwrap();
1368            for forbidden in [
1369                "request_body",
1370                "response_body",
1371                "cookie",
1372                "session",
1373                "token",
1374                "password",
1375                "connection_string",
1376                "db_value",
1377            ] {
1378                assert!(!encoded.contains(forbidden));
1379            }
1380        }
1381    }
1382
1383    #[test]
1384    fn fixed_metrics_snapshot_tracks_closed_low_cardinality_events() {
1385        let observer = Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap();
1386        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
1387        let common = event_context(1);
1388        observer
1389            .start_stage(root.context(), Stage::Response, event_context(1))
1390            .succeed();
1391        observer.record_capacity(
1392            root.context(),
1393            &common,
1394            CapacityObservation::rejected_dimension(
1395                CapacityDimension::Database,
1396                100,
1397                80,
1398                80,
1399                BottleneckIdentity::new("database_slots").unwrap(),
1400                RejectReason::new("limit_reached").unwrap(),
1401                2,
1402            ),
1403        );
1404        observer.record_database_disposition(
1405            root.context(),
1406            &common,
1407            DatabaseDisposition::Discarded,
1408        );
1409        observer.record_outbound(
1410            root.context(),
1411            &common,
1412            OutboundObservation::new(
1413                RouteIdentity::new("cn-hz-a").unwrap(),
1414                OutboundAuthority::new("inventory-service").unwrap(),
1415                OutboundResult::Timeout,
1416            ),
1417        );
1418        observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
1419        observer.record_lifecycle(
1420            root.context(),
1421            &common,
1422            LifecycleState::Running,
1423            Health::Healthy,
1424        );
1425
1426        let snapshot = observer.metrics_snapshot();
1427        assert_eq!(snapshot.requests(StageOutcome::Success), 1);
1428        assert_eq!(
1429            snapshot.stage_latency(Stage::Response).iter().sum::<u64>(),
1430            1
1431        );
1432        assert_eq!(
1433            snapshot.capacity_rejected(Some(CapacityDimension::Database)),
1434            1
1435        );
1436        assert_eq!(
1437            snapshot.capacity_used(Some(CapacityDimension::Database)),
1438            80
1439        );
1440        assert_eq!(snapshot.database(DatabaseDisposition::Discarded), 1);
1441        assert_eq!(snapshot.outbound(OutboundResult::Timeout), 1);
1442        assert_eq!(snapshot.logger_dropped(), 0);
1443        assert_eq!(snapshot.logger_output_failed(), 1);
1444        assert_eq!(snapshot.lifecycle(), LifecycleState::Running);
1445        assert_eq!(snapshot.health(), Health::Healthy);
1446        assert!(snapshot.ready());
1447        root.succeed();
1448    }
1449
1450    #[test]
1451    fn typed_capacity_and_resource_terminal_preserve_closed_schema() {
1452        let capture = Capture::default();
1453        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
1454        let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
1455        let common = event_context(1);
1456        observer.record_capacity(
1457            root.context(),
1458            &common,
1459            CapacityObservation::rejected_dimension(
1460                CapacityDimension::Database,
1461                4,
1462                2,
1463                2,
1464                BottleneckIdentity::new("database").unwrap(),
1465                RejectReason::new("at_limit").unwrap(),
1466                17,
1467            ),
1468        );
1469        observer.record_resource_finalization(
1470            root.context(),
1471            &common,
1472            DatabaseDisposition::NotUsed,
1473            23,
1474        );
1475        root.succeed();
1476        block_on(observer.flush()).unwrap();
1477
1478        let records = records(&capture);
1479        let capacity = records
1480            .iter()
1481            .find(|record| record["event"] == "framework.capacity")
1482            .unwrap();
1483        assert_eq!(capacity["capacity_dimension"], "database");
1484        assert_eq!(capacity["budget"], 4);
1485        assert_eq!(capacity["limit"], 2);
1486        assert_eq!(capacity["used"], 2);
1487        assert_eq!(capacity["bottleneck"], "database");
1488        assert_eq!(capacity["reject_reason"], "at_limit");
1489        assert_eq!(capacity["elapsed_ms"], 17);
1490
1491        let terminal = records
1492            .iter()
1493            .find(|record| record["event"] == "framework.resource.finalized")
1494            .unwrap();
1495        assert_eq!(terminal["stage"], "resource_finalization");
1496        assert_eq!(terminal["credit"], "released");
1497        assert_eq!(terminal["db_disposition"], "not_used");
1498        assert_eq!(terminal["elapsed_ms"], 23);
1499        assert_eq!(terminal["outcome"], "success");
1500        assert_eq!(capacity["trace_id"], terminal["trace_id"]);
1501        assert_eq!(capacity["rpc_id"], terminal["rpc_id"]);
1502        assert_eq!(capacity["request_identity"], terminal["request_identity"]);
1503        assert_eq!(capacity["route"], terminal["route"]);
1504        assert_eq!(capacity["attempt"], terminal["attempt"]);
1505    }
1506
1507    #[test]
1508    fn unsafe_or_unbounded_identifiers_and_zero_attempt_are_rejected() {
1509        assert_eq!(RequestIdentity::new(""), Err(ChainFieldError::Empty));
1510        assert_eq!(
1511            RouteIdentity::new("x".repeat(257)),
1512            Err(ChainFieldError::TooLong)
1513        );
1514        assert_eq!(
1515            RejectReason::new("bad\nreason"),
1516            Err(ChainFieldError::ControlCharacter)
1517        );
1518        for unsafe_value in ["https://user:secret@host/path", "host/path", "host?token=x"] {
1519            assert_eq!(
1520                OutboundAuthority::new(unsafe_value),
1521                Err(ChainFieldError::UnsafeAuthority)
1522            );
1523        }
1524        assert_eq!(
1525            EventContext::new(
1526                RequestIdentity::new("r").unwrap(),
1527                RouteIdentity::new("route").unwrap(),
1528                0
1529            ),
1530            Err(ChainFieldError::ZeroAttempt),
1531        );
1532    }
1533
1534    #[test]
1535    fn borrowed_event_identity_preserves_utf8_limit_without_owned_heap() {
1536        let full = format!("{}a", "界".repeat(85));
1537        assert_eq!(full.len(), MAX_IDENTITY_BYTES);
1538        let event = EventContext::new(
1539            RequestIdentity::from_borrowed(&full).unwrap(),
1540            RouteIdentity::from_borrowed("quote.route").unwrap(),
1541            1,
1542        ).unwrap();
1543        assert_eq!(event.request.as_str(), full);
1544        assert_eq!(event.clone(), event);
1545        assert!(!std::mem::needs_drop::<EventContext>());
1546        assert_eq!(RequestIdentity::from_borrowed("bad\nrequest"), Err(ChainFieldError::ControlCharacter));
1547    }
1548}
saddle_observability/chain.rs

saddle_observability/
chain.rs
