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
50safe_identity!(RequestIdentity);
51safe_identity!(RouteIdentity);
52safe_identity!(BottleneckIdentity);
53safe_identity!(RejectReason);
54
55#[derive(Clone, Debug, Eq, PartialEq)]
56pub struct OutboundAuthority(String);
57
58impl OutboundAuthority {
59 pub fn new(value: impl Into<String>) -> Result<Self, ChainFieldError> {
60 let value = validate_identity(value.into())?;
61 if value.contains(['@', '?', '#', '/', '\\']) || value.contains("://") {
62 return Err(ChainFieldError::UnsafeAuthority);
63 }
64 Ok(Self(value))
65 }
66}
67
68fn validate_identity(value: String) -> Result<String, ChainFieldError> {
69 if value.is_empty() {
70 return Err(ChainFieldError::Empty);
71 }
72 if value.len() > MAX_IDENTITY_BYTES {
73 return Err(ChainFieldError::TooLong);
74 }
75 if value.chars().any(char::is_control) {
76 return Err(ChainFieldError::ControlCharacter);
77 }
78 Ok(value)
79}
80
81#[derive(Clone, Copy, Debug, Eq, PartialEq)]
82pub enum Stage {
83 Ingress,
84 Admission,
85 Handler,
86 Database,
87 ProfuseContract,
88 Response,
89 ResourceFinalization,
90}
91
92impl Stage {
93 const fn as_str(self) -> &'static str {
94 match self {
95 Self::Ingress => "ingress",
96 Self::Admission => "admission",
97 Self::Handler => "handler",
98 Self::Database => "database",
99 Self::ProfuseContract => "profuse_contract",
100 Self::Response => "response",
101 Self::ResourceFinalization => "resource_finalization",
102 }
103 }
104}
105
106#[derive(Clone, Copy, Debug, Eq, PartialEq)]
107pub enum StageOutcome {
108 Success,
109 Failure,
110 Rejected,
111 Cancelled,
112}
113
114impl StageOutcome {
115 const fn as_str(self) -> &'static str {
116 match self {
117 Self::Success => "success",
118 Self::Failure => "failure",
119 Self::Rejected => "rejected",
120 Self::Cancelled => "cancelled",
121 }
122 }
123}
124
125#[derive(Clone, Debug, Eq, PartialEq)]
126pub struct EventContext {
127 request: RequestIdentity,
128 route: RouteIdentity,
129 attempt: u32,
130}
131
132impl EventContext {
133 pub fn new(
134 request: RequestIdentity,
135 route: RouteIdentity,
136 attempt: u32,
137 ) -> Result<Self, ChainFieldError> {
138 if attempt == 0 {
139 return Err(ChainFieldError::ZeroAttempt);
140 }
141 Ok(Self {
142 request,
143 route,
144 attempt,
145 })
146 }
147}
148
149pub struct ActiveStage {
150 observer: Observer,
151 context: CallContext,
152 parent_rpc_id: String,
153 stage: Stage,
154 request: RequestIdentity,
155 route: RouteIdentity,
156 attempt: u32,
157 started_at: Instant,
158 finished: bool,
159}
160
161impl Observer {
162 pub fn start_stage(
163 &self,
164 parent: &CallContext,
165 stage: Stage,
166 event_context: EventContext,
167 ) -> ActiveStage {
168 let context = CallContext::new(
169 parent.application().clone(),
170 parent.module().clone(),
171 parent.service().clone(),
172 parent.operation().clone(),
173 parent.trace_id(),
174 self.new_span_id(),
175 )
176 .with_trace_correlation_id(parent.trace_correlation_id().clone());
177 let active = ActiveStage {
178 observer: self.clone(),
179 context,
180 parent_rpc_id: parent.span_id().to_string(),
181 stage,
182 request: event_context.request,
183 route: event_context.route,
184 attempt: event_context.attempt,
185 started_at: Instant::now(),
186 finished: false,
187 };
188 active.emit_started();
189 active
190 }
191
192 pub fn record_capacity(
193 &self,
194 context: &CallContext,
195 event_context: &EventContext,
196 value: CapacityObservation,
197 ) {
198 let mut record = base_record(context, EventLevel::Info, "framework.capacity", "admission");
199 add_event_context(&mut record, event_context);
200 if let Some(dimension) = value.dimension {
201 record.data.insert(
202 "capacity_dimension".into(),
203 Value::String(dimension.as_str().into()),
204 );
205 }
206 record.data.insert("budget".into(), json!(value.budget));
207 record.data.insert("limit".into(), json!(value.limit));
208 record.data.insert("used".into(), json!(value.used));
209 record
210 .data
211 .insert("elapsed_ms".into(), json!(value.elapsed_ms));
212 record
213 .data
214 .insert("bottleneck".into(), Value::String(value.bottleneck.0));
215 record.data.insert(
216 "outcome".into(),
217 Value::String(
218 if value.reject_reason.is_some() {
219 "rejected"
220 } else {
221 "accepted"
222 }
223 .into(),
224 ),
225 );
226 if let Some(reason) = value.reject_reason {
227 record
228 .data
229 .insert("reject_reason".into(), Value::String(reason.0));
230 }
231 self.emit(record);
232 }
233
234 pub fn record_database_disposition(
235 &self,
236 context: &CallContext,
237 event_context: &EventContext,
238 disposition: DatabaseDisposition,
239 ) {
240 let mut record = base_record(
241 context,
242 EventLevel::Info,
243 "framework.database.disposition",
244 "database",
245 );
246 add_event_context(&mut record, event_context);
247 record.data.insert(
248 "db_disposition".into(),
249 Value::String(disposition.as_str().into()),
250 );
251 record
252 .data
253 .insert("outcome".into(), Value::String(disposition.as_str().into()));
254 self.emit(record);
255 }
256
257 pub fn record_resource_finalization(
258 &self,
259 context: &CallContext,
260 event_context: &EventContext,
261 disposition: DatabaseDisposition,
262 elapsed_ms: u64,
263 ) {
264 let mut record = base_record(
265 context,
266 EventLevel::Info,
267 "framework.resource.finalized",
268 "resource_finalization",
269 );
270 add_event_context(&mut record, event_context);
271 record.data.insert("elapsed_ms".into(), json!(elapsed_ms));
272 record
273 .data
274 .insert("credit".into(), Value::String("released".into()));
275 record.data.insert(
276 "db_disposition".into(),
277 Value::String(disposition.as_str().into()),
278 );
279 record
280 .data
281 .insert("outcome".into(), Value::String("success".into()));
282 self.emit(record);
283 }
284
285 pub fn record_outbound(
286 &self,
287 context: &CallContext,
288 event_context: &EventContext,
289 observation: OutboundObservation,
290 ) {
291 let mut record = base_record(
292 context,
293 EventLevel::Info,
294 "framework.outbound",
295 "profuse_contract",
296 );
297 add_event_context(&mut record, event_context);
298 record
299 .data
300 .insert("zone".into(), Value::String(observation.zone.0));
301 record
302 .data
303 .insert("authority".into(), Value::String(observation.authority.0));
304 record.data.insert(
305 "outbound_result".into(),
306 Value::String(observation.result.as_str().into()),
307 );
308 record.data.insert(
309 "outcome".into(),
310 Value::String(observation.result.as_str().into()),
311 );
312 self.emit(record);
313 }
314
315 pub fn record_lifecycle(
316 &self,
317 context: &CallContext,
318 event_context: &EventContext,
319 state: LifecycleState,
320 health: Health,
321 ) {
322 let mut record = base_record(
323 context,
324 if health == Health::Healthy {
325 EventLevel::Info
326 } else {
327 EventLevel::Error
328 },
329 "framework.lifecycle",
330 "resource_finalization",
331 );
332 add_event_context(&mut record, event_context);
333 record
334 .data
335 .insert("lifecycle".into(), Value::String(state.as_str().into()));
336 record
337 .data
338 .insert("health".into(), Value::String(health.as_str().into()));
339 record
340 .data
341 .insert("outcome".into(), Value::String(health.as_str().into()));
342 self.emit(record);
343 }
344
345 pub fn record_logger_health(
346 &self,
347 context: &CallContext,
348 event_context: &EventContext,
349 dropped: u64,
350 failure: Option<OutputStage>,
351 ) {
352 let mut record = base_record(
353 context,
354 if failure.is_some() || dropped > 0 {
355 EventLevel::Error
356 } else {
357 EventLevel::Info
358 },
359 "framework.logger.health",
360 "logger",
361 );
362 add_event_context(&mut record, event_context);
363 record.data.insert("logger_dropped".into(), json!(dropped));
364 record.data.insert(
365 "logger_health".into(),
366 Value::String(
367 if failure.is_some() {
368 "output_failed"
369 } else {
370 "healthy"
371 }
372 .into(),
373 ),
374 );
375 record.data.insert(
376 "outcome".into(),
377 Value::String(
378 if failure.is_some() {
379 "failure"
380 } else {
381 "success"
382 }
383 .into(),
384 ),
385 );
386 if let Some(stage) = failure {
387 record.data.insert(
388 "output_failed".into(),
389 Value::String(format!("{stage:?}").to_ascii_lowercase()),
390 );
391 }
392 self.emit(record);
393 }
394}
395
396impl ActiveStage {
397 pub fn context(&self) -> &CallContext {
398 &self.context
399 }
400 pub fn succeed(mut self) {
401 self.finish(StageOutcome::Success, None);
402 }
403 pub fn reject(mut self, error: &SaddleError) {
404 self.finish(StageOutcome::Rejected, Some(error));
405 }
406 pub fn fail(mut self, error: &SaddleError) {
407 self.finish(StageOutcome::Failure, Some(error));
408 }
409
410 fn emit_started(&self) {
411 let mut record = self.record(EventLevel::Info, "framework.stage.started");
412 record
413 .data
414 .insert("outcome".into(), Value::String("started".into()));
415 self.observer.emit(record);
416 }
417
418 fn finish(&mut self, outcome: StageOutcome, error: Option<&SaddleError>) {
419 if self.finished {
420 return;
421 }
422 let mut record = self.record(
423 if outcome == StageOutcome::Success {
424 EventLevel::Info
425 } else {
426 EventLevel::Error
427 },
428 "framework.stage.finished",
429 );
430 record
431 .data
432 .insert("outcome".into(), Value::String(outcome.as_str().into()));
433 record.data.insert(
434 "elapsed_ms".into(),
435 json!(u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX)),
436 );
437 if let Some(error) = error {
438 record
439 .data
440 .insert("error_code".into(), Value::String(error.code().to_owned()));
441 record.data.insert(
442 "error_kind".into(),
443 Value::String(error_kind(error.kind()).into()),
444 );
445 }
446 self.observer.emit(record);
447 self.finished = true;
448 }
449
450 fn record(&self, level: EventLevel, event: &'static str) -> LogRecord {
451 let mut record = base_record(&self.context, level, event, self.stage.as_str());
452 record.parent = Some(self.parent_rpc_id.clone());
453 record.parent_span_id = Some(self.parent_rpc_id.clone());
454 record.data.insert(
455 "request_identity".into(),
456 Value::String(self.request.0.clone()),
457 );
458 record
459 .data
460 .insert("route".into(), Value::String(self.route.0.clone()));
461 record.data.insert("attempt".into(), json!(self.attempt));
462 record.data.insert("elapsed_ms".into(), json!(0));
463 record
464 .data
465 .insert("error_code".into(), Value::String("none".into()));
466 record
467 }
468}
469
470impl Drop for ActiveStage {
471 fn drop(&mut self) {
472 self.finish(StageOutcome::Cancelled, None);
473 }
474}
475
476pub struct CapacityObservation {
477 dimension: Option<CapacityDimension>,
478 budget: u64,
479 limit: u64,
480 used: u64,
481 bottleneck: BottleneckIdentity,
482 reject_reason: Option<RejectReason>,
483 elapsed_ms: u64,
484}
485
486impl CapacityObservation {
487 pub fn accepted(budget: u64, limit: u64, used: u64, bottleneck: BottleneckIdentity) -> Self {
488 Self {
489 dimension: None,
490 budget,
491 limit,
492 used,
493 bottleneck,
494 reject_reason: None,
495 elapsed_ms: 0,
496 }
497 }
498 pub fn rejected(
499 budget: u64,
500 limit: u64,
501 used: u64,
502 bottleneck: BottleneckIdentity,
503 reason: RejectReason,
504 ) -> Self {
505 Self {
506 dimension: None,
507 budget,
508 limit,
509 used,
510 bottleneck,
511 reject_reason: Some(reason),
512 elapsed_ms: 0,
513 }
514 }
515
516 pub fn accepted_dimension(
517 dimension: CapacityDimension,
518 budget: u64,
519 limit: u64,
520 used: u64,
521 bottleneck: BottleneckIdentity,
522 elapsed_ms: u64,
523 ) -> Self {
524 Self {
525 dimension: Some(dimension),
526 budget,
527 limit,
528 used,
529 bottleneck,
530 reject_reason: None,
531 elapsed_ms,
532 }
533 }
534
535 pub fn rejected_dimension(
536 dimension: CapacityDimension,
537 budget: u64,
538 limit: u64,
539 used: u64,
540 bottleneck: BottleneckIdentity,
541 reason: RejectReason,
542 elapsed_ms: u64,
543 ) -> Self {
544 Self {
545 dimension: Some(dimension),
546 budget,
547 limit,
548 used,
549 bottleneck,
550 reject_reason: Some(reason),
551 elapsed_ms,
552 }
553 }
554}
555
556#[derive(Clone, Copy, Debug, Eq, PartialEq)]
557pub enum CapacityDimension {
558 Cpu,
559 Memory,
560 Database,
561 ProfuseContract,
562}
563
564impl CapacityDimension {
565 const fn as_str(self) -> &'static str {
566 match self {
567 Self::Cpu => "cpu",
568 Self::Memory => "memory",
569 Self::Database => "database",
570 Self::ProfuseContract => "profuse_contract",
571 }
572 }
573}
574
575#[derive(Clone, Copy, Debug, Eq, PartialEq)]
576pub enum DatabaseDisposition {
577 NotUsed,
578 Returned,
579 Discarded,
580}
581impl DatabaseDisposition {
582 const fn as_str(self) -> &'static str {
583 match self {
584 Self::NotUsed => "not_used",
585 Self::Returned => "returned",
586 Self::Discarded => "discarded",
587 }
588 }
589}
590
591pub struct OutboundObservation {
592 zone: RouteIdentity,
593 authority: OutboundAuthority,
594 result: OutboundResult,
595}
596impl OutboundObservation {
597 pub fn new(zone: RouteIdentity, authority: OutboundAuthority, result: OutboundResult) -> Self {
598 Self {
599 zone,
600 authority,
601 result,
602 }
603 }
604}
605
606#[derive(Clone, Copy, Debug, Eq, PartialEq)]
607pub enum OutboundResult {
608 Success,
609 Failure,
610 Rejected,
611 Timeout,
612}
613impl OutboundResult {
614 const fn as_str(self) -> &'static str {
615 match self {
616 Self::Success => "success",
617 Self::Failure => "failure",
618 Self::Rejected => "rejected",
619 Self::Timeout => "timeout",
620 }
621 }
622}
623
624#[derive(Clone, Copy, Debug, Eq, PartialEq)]
625pub enum LifecycleState {
626 Starting,
627 Running,
628 Draining,
629 Stopped,
630}
631impl LifecycleState {
632 const fn as_str(self) -> &'static str {
633 match self {
634 Self::Starting => "starting",
635 Self::Running => "running",
636 Self::Draining => "draining",
637 Self::Stopped => "stopped",
638 }
639 }
640}
641
642#[derive(Clone, Copy, Debug, Eq, PartialEq)]
643pub enum Health {
644 Healthy,
645 Degraded,
646 Failed,
647}
648impl Health {
649 const fn as_str(self) -> &'static str {
650 match self {
651 Self::Healthy => "healthy",
652 Self::Degraded => "degraded",
653 Self::Failed => "failed",
654 }
655 }
656}
657
658fn base_record(
659 context: &CallContext,
660 level: EventLevel,
661 event: &'static str,
662 stage: &'static str,
663) -> LogRecord {
664 let mut record = LogRecord::new(level, event);
665 record.trace_id = Some(context.trace_correlation_id().to_string());
666 record.span = Some(stage.into());
667 record.span_id = Some(context.span_id().to_string());
668 record
669 .data
670 .insert("timestamp".into(), json!(record.timestamp_unix_ms));
671 record
672 .data
673 .insert("stage".into(), Value::String(stage.into()));
674 record.data.insert(
675 "rpc_id".into(),
676 Value::String(context.span_id().to_string()),
677 );
678 record.data.insert(
679 "request".into(),
680 Value::String(context.operation().to_string()),
681 );
682 record
683}
684
685fn add_event_context(record: &mut LogRecord, context: &EventContext) {
686 record.data.insert(
687 "request_identity".into(),
688 Value::String(context.request.0.clone()),
689 );
690 record
691 .data
692 .insert("route".into(), Value::String(context.route.0.clone()));
693 record.data.insert("attempt".into(), json!(context.attempt));
694 record.data.insert("elapsed_ms".into(), json!(0));
695 record
696 .data
697 .insert("error_code".into(), Value::String("none".into()));
698}
699
700const fn error_kind(kind: ErrorKind) -> &'static str {
701 match kind {
702 ErrorKind::InvalidArgument => "invalid_argument",
703 ErrorKind::NotFound => "not_found",
704 ErrorKind::Conflict => "conflict",
705 ErrorKind::Business => "business",
706 ErrorKind::Unavailable => "unavailable",
707 ErrorKind::Infrastructure => "infrastructure",
708 ErrorKind::Internal => "internal",
709 _ => "unknown",
710 }
711}
712
713#[cfg(test)]
714mod tests {
715 use std::{
716 future::Future,
717 io,
718 sync::{Arc, Mutex},
719 task::{Context, Poll, Wake, Waker},
720 thread,
721 };
722
723 use super::*;
724 use crate::ObserverConfig;
725
726 #[derive(Clone, Default)]
727 struct Capture(Arc<Mutex<Vec<u8>>>);
728
729 impl io::Write for Capture {
730 fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
731 self.0.lock().unwrap().extend_from_slice(bytes);
732 Ok(bytes.len())
733 }
734 fn flush(&mut self) -> io::Result<()> {
735 Ok(())
736 }
737 }
738
739 struct ThreadWaker(thread::Thread);
740 impl Wake for ThreadWaker {
741 fn wake(self: Arc<Self>) {
742 self.0.unpark();
743 }
744 }
745
746 fn block_on<T>(future: impl Future<Output = T>) -> T {
747 let waker = Waker::from(Arc::new(ThreadWaker(thread::current())));
748 let mut context = Context::from_waker(&waker);
749 let mut future = std::pin::pin!(future);
750 loop {
751 match future.as_mut().poll(&mut context) {
752 Poll::Ready(output) => return output,
753 Poll::Pending => thread::park(),
754 }
755 }
756 }
757
758 fn event_context(attempt: u32) -> EventContext {
759 EventContext::new(
760 RequestIdentity::new("request-7").unwrap(),
761 RouteIdentity::new("orders.create").unwrap(),
762 attempt,
763 )
764 .unwrap()
765 }
766
767 fn records(capture: &Capture) -> Vec<Value> {
768 String::from_utf8(capture.0.lock().unwrap().clone())
769 .unwrap()
770 .lines()
771 .map(|line| serde_json::from_str(line).unwrap())
772 .collect()
773 }
774
775 #[test]
776 fn stage_chain_is_correlated_ordered_and_has_one_terminal() {
777 let capture = Capture::default();
778 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
779 let (root, _) = observer.start_external_call(
780 "shop",
781 "entry",
782 "orders",
783 "create",
784 Some("00112233445566778899aabbccddeeff"),
785 );
786 let stages = [
787 Stage::Ingress,
788 Stage::Admission,
789 Stage::Handler,
790 Stage::Database,
791 Stage::ProfuseContract,
792 Stage::Response,
793 Stage::ResourceFinalization,
794 ];
795 for (index, stage) in stages.into_iter().enumerate() {
796 observer
797 .start_stage(root.context(), stage, event_context((index + 1) as u32))
798 .succeed();
799 }
800 drop(observer.start_stage(root.context(), Stage::Handler, event_context(8)));
801 root.succeed();
802 block_on(observer.flush()).unwrap();
803
804 let records = records(&capture);
805 let stage_records: Vec<_> = records
806 .iter()
807 .filter(|value| {
808 value["event"]
809 .as_str()
810 .unwrap()
811 .starts_with("framework.stage.")
812 })
813 .collect();
814 assert_eq!(stage_records.len(), 16);
815 for pair in stage_records.chunks_exact(2) {
816 assert_eq!(pair[0]["event"], "framework.stage.started");
817 assert_eq!(pair[1]["event"], "framework.stage.finished");
818 assert_eq!(pair[0]["rpc_id"], pair[1]["rpc_id"]);
819 assert_eq!(pair[0]["trace_id"], "00112233445566778899aabbccddeeff");
820 for field in [
821 "timestamp_unix_ms",
822 "timestamp",
823 "level",
824 "event",
825 "stage",
826 "trace_id",
827 "rpc_id",
828 "request_identity",
829 "route",
830 "attempt",
831 "outcome",
832 "error_code",
833 ] {
834 assert!(pair[0].get(field).is_some(), "missing {field}");
835 }
836 assert!(pair[1].get("elapsed_ms").is_some());
837 }
838 assert_eq!(stage_records.last().unwrap()["outcome"], "cancelled");
839 }
840
841 #[test]
842 fn closed_observations_have_common_fields_and_no_sensitive_payload() {
843 let capture = Capture::default();
844 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
845 let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
846 let common = event_context(1);
847 observer.record_capacity(
848 root.context(),
849 &common,
850 CapacityObservation::rejected(
851 100,
852 80,
853 80,
854 BottleneckIdentity::new("request_slots").unwrap(),
855 RejectReason::new("limit_reached").unwrap(),
856 ),
857 );
858 observer.record_database_disposition(
859 root.context(),
860 &common,
861 DatabaseDisposition::Returned,
862 );
863 observer.record_outbound(
864 root.context(),
865 &common,
866 OutboundObservation::new(
867 RouteIdentity::new("cn-hz-a").unwrap(),
868 OutboundAuthority::new("inventory-service").unwrap(),
869 OutboundResult::Success,
870 ),
871 );
872 observer.record_lifecycle(
873 root.context(),
874 &common,
875 LifecycleState::Draining,
876 Health::Healthy,
877 );
878 observer.record_logger_health(root.context(), &common, 3, Some(OutputStage::Record));
879 root.succeed();
880 let _ = block_on(observer.flush());
881
882 let records: Vec<_> = records(&capture)
883 .into_iter()
884 .filter(|value| {
885 matches!(
886 value["event"].as_str(),
887 Some(
888 "framework.capacity"
889 | "framework.database.disposition"
890 | "framework.outbound"
891 | "framework.lifecycle"
892 | "framework.logger.health"
893 )
894 )
895 })
896 .collect();
897 assert_eq!(records.len(), 5);
898 for record in records {
899 for field in [
900 "timestamp_unix_ms",
901 "timestamp",
902 "level",
903 "event",
904 "stage",
905 "trace_id",
906 "rpc_id",
907 "request_identity",
908 "route",
909 "attempt",
910 "elapsed_ms",
911 "outcome",
912 "error_code",
913 ] {
914 assert!(record.get(field).is_some(), "missing {field}");
915 }
916 let encoded = serde_json::to_string(&record).unwrap();
917 for forbidden in [
918 "request_body",
919 "response_body",
920 "cookie",
921 "session",
922 "token",
923 "password",
924 "connection_string",
925 "db_value",
926 ] {
927 assert!(!encoded.contains(forbidden));
928 }
929 }
930 }
931
932 #[test]
933 fn typed_capacity_and_resource_terminal_preserve_closed_schema() {
934 let capture = Capture::default();
935 let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
936 let (root, _) = observer.start_external_call("shop", "entry", "orders", "create", None);
937 let common = event_context(1);
938 observer.record_capacity(
939 root.context(),
940 &common,
941 CapacityObservation::rejected_dimension(
942 CapacityDimension::Database,
943 4,
944 2,
945 2,
946 BottleneckIdentity::new("database").unwrap(),
947 RejectReason::new("at_limit").unwrap(),
948 17,
949 ),
950 );
951 observer.record_resource_finalization(
952 root.context(),
953 &common,
954 DatabaseDisposition::NotUsed,
955 23,
956 );
957 root.succeed();
958 block_on(observer.flush()).unwrap();
959
960 let records = records(&capture);
961 let capacity = records
962 .iter()
963 .find(|record| record["event"] == "framework.capacity")
964 .unwrap();
965 assert_eq!(capacity["capacity_dimension"], "database");
966 assert_eq!(capacity["budget"], 4);
967 assert_eq!(capacity["limit"], 2);
968 assert_eq!(capacity["used"], 2);
969 assert_eq!(capacity["bottleneck"], "database");
970 assert_eq!(capacity["reject_reason"], "at_limit");
971 assert_eq!(capacity["elapsed_ms"], 17);
972
973 let terminal = records
974 .iter()
975 .find(|record| record["event"] == "framework.resource.finalized")
976 .unwrap();
977 assert_eq!(terminal["stage"], "resource_finalization");
978 assert_eq!(terminal["credit"], "released");
979 assert_eq!(terminal["db_disposition"], "not_used");
980 assert_eq!(terminal["elapsed_ms"], 23);
981 assert_eq!(terminal["outcome"], "success");
982 assert_eq!(capacity["trace_id"], terminal["trace_id"]);
983 assert_eq!(capacity["rpc_id"], terminal["rpc_id"]);
984 assert_eq!(capacity["request_identity"], terminal["request_identity"]);
985 assert_eq!(capacity["route"], terminal["route"]);
986 assert_eq!(capacity["attempt"], terminal["attempt"]);
987 }
988
989 #[test]
990 fn unsafe_or_unbounded_identifiers_and_zero_attempt_are_rejected() {
991 assert_eq!(RequestIdentity::new(""), Err(ChainFieldError::Empty));
992 assert_eq!(
993 RouteIdentity::new("x".repeat(257)),
994 Err(ChainFieldError::TooLong)
995 );
996 assert_eq!(
997 RejectReason::new("bad\nreason"),
998 Err(ChainFieldError::ControlCharacter)
999 );
1000 for unsafe_value in ["https://user:secret@host/path", "host/path", "host?token=x"] {
1001 assert_eq!(
1002 OutboundAuthority::new(unsafe_value),
1003 Err(ChainFieldError::UnsafeAuthority)
1004 );
1005 }
1006 assert_eq!(
1007 EventContext::new(
1008 RequestIdentity::new("r").unwrap(),
1009 RouteIdentity::new("route").unwrap(),
1010 0
1011 ),
1012 Err(ChainFieldError::ZeroAttempt),
1013 );
1014 }
1015}