1use crate::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext};
4use saddle_core::{
5 BoundedDiagnostic, BoundedDiagnosticCause, CallContext, CaptureSite, DiagnosticCategory,
6 DiagnosticOccurrence, DiagnosticOutcomeAxes,
7};
8use serde::{Serialize, Serializer};
9
10#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
11#[serde(tag = "state", content = "value", rename_all = "snake_case")]
12enum Field<T> {
13 Present(T),
14 NotApplicable,
15 NotEstablished,
16 Unavailable,
17}
18
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
22pub enum DiagnosticContextMissing {
23 NotApplicable,
24 NotEstablished,
25 Unavailable,
26}
27
28#[derive(Clone, Copy)]
31pub struct DiagnosticDbOperation(&'static str);
32impl DiagnosticDbOperation {
33 pub fn from_registered(value: &'static str) -> Option<Self> {
34 (!value.is_empty()
35 && value.len() <= 128
36 && value
37 .bytes()
38 .all(|b| b.is_ascii_alphanumeric() || b"._:-".contains(&b)))
39 .then_some(Self(value))
40 }
41}
42
43#[derive(Clone, Copy)]
46pub struct DiagnosticZone(ProtocolId);
47#[derive(Clone, Copy, Debug, Eq, PartialEq)]
48pub enum DiagnosticZoneError {
49 Empty,
50 TooLong,
51 Unsafe,
52}
53impl DiagnosticZone {
54 pub fn from_validated_ingress(value: &str) -> Result<Self, DiagnosticZoneError> {
56 if value.len() > 256 {
57 return Err(DiagnosticZoneError::TooLong);
58 }
59 if value.trim().is_empty() {
60 return Err(DiagnosticZoneError::Empty);
61 }
62 if value.chars().any(char::is_control)
63 || value.contains("://")
64 || value.contains(['@', '?', '#', '\\'])
65 {
66 return Err(DiagnosticZoneError::Unsafe);
67 }
68 Ok(Self(ProtocolId::copy(value)))
69 }
70}
71
72#[derive(Clone, Copy, Eq, PartialEq)]
73struct SafeText {
74 value: ProtocolId,
75 truncated: bool,
76 redacted: bool,
77}
78impl SafeText {
79 fn metadata(value: &str) -> Self {
80 let mut len = value.len().min(256);
81 while !value.is_char_boundary(len) {
82 len -= 1;
83 }
84 let prefix = &value[..len];
85 let redacted = prefix.contains("://")
86 || prefix
87 .chars()
88 .any(|c| !(c.is_alphanumeric() || "_./:{}*-".contains(c)));
89 Self {
90 value: ProtocolId::copy(if redacted { "" } else { prefix }),
91 truncated: len < value.len(),
92 redacted,
93 }
94 }
95}
96impl Serialize for SafeText {
97 fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
98 use serde::ser::SerializeStruct;
99 let mut out = s.serialize_struct("SafeText", 3)?;
100 out.serialize_field("value", &self.value)?;
101 out.serialize_field("truncated", &self.truncated)?;
102 out.serialize_field("redacted", &self.redacted)?;
103 out.end()
104 }
105}
106#[derive(Clone, Copy, Eq, PartialEq)]
107struct Span(u64);
108impl Serialize for Span {
109 fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
110 let mut bytes = [b'0'; 16];
111 for (i, b) in bytes.iter_mut().enumerate() {
112 *b = b"0123456789abcdef"[((self.0 >> ((15 - i) * 4)) & 15) as usize];
113 }
114 s.serialize_str(std::str::from_utf8(&bytes).unwrap())
115 }
116}
117
118#[derive(Clone, Copy, Eq, PartialEq)]
120struct ProtocolId {
121 bytes: [u8; 256],
122 len: usize,
123}
124impl ProtocolId {
125 fn copy(value: &str) -> Self {
126 let mut bytes = [0; 256];
128 bytes[..value.len()].copy_from_slice(value.as_bytes());
129 Self {
130 bytes,
131 len: value.len(),
132 }
133 }
134}
135impl Serialize for ProtocolId {
136 fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
137 s.serialize_str(std::str::from_utf8(&self.bytes[..self.len]).unwrap_or(""))
138 }
139}
140
141#[derive(Clone, Copy, Serialize)]
143struct Projection {
144 schema_version: u8,
145 application: Field<SafeText>,
146 module: Field<SafeText>,
147 service: Field<SafeText>,
148 operation: Field<SafeText>,
149 db_operation: Field<&'static str>,
150 trace_id: Field<ProtocolId>,
151 rpc_id: Field<ProtocolId>,
152 span_id: Field<Span>,
153 request: Field<SafeText>,
154 #[serde(skip)]
157 request_binding: Option<ProtocolId>,
158 route: Field<SafeText>,
159 attempt: Field<u32>,
160 scope: Field<saddle_core::DbScopeDiagnosticIdentity>,
161 task: Field<SafeText>,
162 lifecycle: Field<SafeText>,
163 zone: Field<ProtocolId>,
164 target: Field<SafeText>,
165}
166impl Projection {
167 fn missing(reason: DiagnosticContextMissing) -> Self {
168 fn absent<T>(reason: DiagnosticContextMissing) -> Field<T> {
169 match reason {
170 DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
171 DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
172 DiagnosticContextMissing::Unavailable => Field::Unavailable,
173 }
174 }
175 Self {
176 schema_version: 1,
177 application: absent(reason),
178 module: absent(reason),
179 service: absent(reason),
180 operation: absent(reason),
181 db_operation: absent(reason),
182 trace_id: absent(reason),
183 rpc_id: absent(reason),
184 span_id: absent(reason),
185 request: absent(reason),
186 request_binding: None,
187 route: absent(reason),
188 attempt: absent(reason),
189 scope: absent(reason),
190 task: absent(reason),
191 lifecycle: absent(reason),
192 zone: absent(reason),
193 target: absent(reason),
194 }
195 }
196 fn existing(call: &CallContext, event: &EventContext) -> Self {
197 let text = |s: &str| Field::Present(SafeText::metadata(s));
198 Self {
199 schema_version: 1,
200 application: text(call.application().as_str()),
201 module: text(call.module().as_str()),
202 service: text(call.service().as_str()),
203 operation: text(call.operation().as_str()),
204 db_operation: Field::Unavailable,
205 trace_id: Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str())),
206 rpc_id: call.rpc_correlation_id().map_or(Field::Unavailable, |id| {
207 Field::Present(ProtocolId::copy(id.as_str()))
208 }),
209 span_id: Field::Present(Span(call.span_id().as_u64())),
210 request: text(event.diagnostic_request()),
211 request_binding: Some(ProtocolId::copy(event.diagnostic_request())),
212 route: text(event.diagnostic_route()),
213 attempt: Field::Present(event.diagnostic_attempt()),
214 scope: Field::Unavailable,
217 task: Field::Unavailable,
218 lifecycle: Field::Unavailable,
219 zone: Field::Unavailable,
220 target: Field::Unavailable,
221 }
222 }
223}
224
225#[derive(Clone, Copy, Debug, Eq, PartialEq)]
227pub enum DiagnosticContextField {
228 Application,
229 Module,
230 Service,
231 Operation,
232 DbOperation,
233 Trace,
234 Rpc,
235 Span,
236 Request,
237 Route,
238 Attempt,
239 Scope,
240 Task,
241 Lifecycle,
242 Zone,
243 Target,
244}
245
246#[must_use = "recover the original context; do not silently discard known identity"]
248pub struct ContextBindingError<T> {
249 field: DiagnosticContextField,
250 original: T,
251}
252impl<T> ContextBindingError<T> {
253 pub fn field(&self) -> DiagnosticContextField {
254 self.field
255 }
256 pub fn into_original(self) -> T {
257 self.original
258 }
259}
260fn known<T: Copy + Eq>(
261 field: &mut Field<T>,
262 value: T,
263 name: DiagnosticContextField,
264) -> Result<(), DiagnosticContextField> {
265 if let Field::Present(old) = field
266 && *old != value
267 {
268 return Err(name);
269 }
270 *field = Field::Present(value);
271 Ok(())
272}
273fn bind_call(context: &mut Projection, call: &CallContext) -> Result<(), DiagnosticContextField> {
274 use DiagnosticContextField as F;
275 known(
276 &mut context.application,
277 SafeText::metadata(call.application().as_str()),
278 F::Application,
279 )?;
280 known(
281 &mut context.module,
282 SafeText::metadata(call.module().as_str()),
283 F::Module,
284 )?;
285 known(
286 &mut context.service,
287 SafeText::metadata(call.service().as_str()),
288 F::Service,
289 )?;
290 known(
291 &mut context.operation,
292 SafeText::metadata(call.operation().as_str()),
293 F::Operation,
294 )?;
295 known(
296 &mut context.trace_id,
297 ProtocolId::copy(call.trace_correlation_id().as_str()),
298 F::Trace,
299 )?;
300 known(&mut context.span_id, Span(call.span_id().as_u64()), F::Span)?;
301 if let Some(rpc) = call.rpc_correlation_id() {
302 known(&mut context.rpc_id, ProtocolId::copy(rpc.as_str()), F::Rpc)?;
303 }
304 Ok(())
305}
306fn bind_event(
307 context: &mut Projection,
308 event: &EventContext,
309) -> Result<(), DiagnosticContextField> {
310 use DiagnosticContextField as F;
311 bind_request(context, event.diagnostic_request())?;
312 known(
313 &mut context.route,
314 SafeText::metadata(event.diagnostic_route()),
315 F::Route,
316 )?;
317 known(&mut context.attempt, event.diagnostic_attempt(), F::Attempt)
318}
319fn bind_request(context: &mut Projection, request: &str) -> Result<(), DiagnosticContextField> {
320 let exact = ProtocolId::copy(request);
321 if context.request_binding.is_some_and(|old| old != exact) {
322 return Err(DiagnosticContextField::Request);
323 }
324 known(
325 &mut context.request,
326 SafeText::metadata(request),
327 DiagnosticContextField::Request,
328 )?;
329 context.request_binding = Some(exact);
330 Ok(())
331}
332
333#[derive(Clone, Copy)]
335pub enum DiagnosticRequestPhase {
336 SocketAccepted,
337 Admission,
338 ReadingHead,
339 ReadingBody,
340 Validation,
341 Dispatch,
342 Response,
343 TaskJoin,
344 Finalization,
345}
346impl DiagnosticRequestPhase {
347 fn as_str(self) -> &'static str {
348 match self {
349 Self::SocketAccepted => "socket_accepted",
350 Self::Admission => "admission",
351 Self::ReadingHead => "reading_head",
352 Self::ReadingBody => "reading_body",
353 Self::Validation => "validation",
354 Self::Dispatch => "dispatch",
355 Self::Response => "response",
356 Self::TaskJoin => "task_join",
357 Self::Finalization => "finalization",
358 }
359 }
360}
361
362#[derive(Clone, Copy)]
366pub struct DiagnosticTaskId(ProtocolId);
367impl DiagnosticTaskId {
368 pub fn from_runtime_id(value: &str) -> Option<Self> {
369 (!value.is_empty() && value.len() <= 64 && value.bytes().all(|b| b.is_ascii_digit()))
370 .then(|| Self(ProtocolId::copy(value)))
371 }
372}
373
374pub struct EarlyRequestContext {
380 context: Projection,
381}
382#[allow(clippy::result_large_err)]
384impl EarlyRequestContext {
385 pub fn socket_accepted(application: &str) -> Self {
388 let mut context = Projection::missing(DiagnosticContextMissing::NotEstablished);
389 context.application = Field::Present(SafeText::metadata(application));
390 context.lifecycle = Field::Present(SafeText::metadata("socket_accepted"));
391 Self { context }
392 }
393 pub fn unavailable() -> Self {
396 Self {
397 context: Projection::missing(DiagnosticContextMissing::Unavailable),
398 }
399 }
400 fn update(
401 mut self,
402 bind: impl FnOnce(&mut Projection) -> Result<(), DiagnosticContextField>,
403 ) -> Result<Self, ContextBindingError<Self>> {
404 let mut next = self.context;
405 if let Err(field) = bind(&mut next) {
406 return Err(ContextBindingError {
407 field,
408 original: self,
409 });
410 }
411 self.context = next;
412 Ok(self)
413 }
414 pub fn with_application(
415 self,
416 application: &saddle_core::ApplicationId,
417 ) -> Result<Self, ContextBindingError<Self>> {
418 self.update(|c| {
419 known(
420 &mut c.application,
421 SafeText::metadata(application.as_str()),
422 DiagnosticContextField::Application,
423 )
424 })
425 }
426 pub fn with_trace(
428 self,
429 trace: &saddle_core::TraceCorrelationId,
430 ) -> Result<Self, ContextBindingError<Self>> {
431 self.update(|c| {
432 known(
433 &mut c.trace_id,
434 ProtocolId::copy(trace.as_str()),
435 DiagnosticContextField::Trace,
436 )
437 })
438 }
439 pub fn with_rpc(
440 self,
441 rpc: &saddle_core::RpcCorrelationId,
442 ) -> Result<Self, ContextBindingError<Self>> {
443 self.update(|c| {
444 known(
445 &mut c.rpc_id,
446 ProtocolId::copy(rpc.as_str()),
447 DiagnosticContextField::Rpc,
448 )
449 })
450 }
451 pub fn with_event(self, event: &EventContext) -> Result<Self, ContextBindingError<Self>> {
452 self.update(|c| bind_event(c, event))
453 }
454 pub fn with_call(self, call: &CallContext) -> Result<Self, ContextBindingError<Self>> {
455 self.update(|c| bind_call(c, call))
456 }
457 pub fn with_request(
458 self,
459 request: &crate::RequestIdentity,
460 ) -> Result<Self, ContextBindingError<Self>> {
461 self.update(|c| bind_request(c, request.as_str()))
462 }
463 pub fn with_route(
464 self,
465 route: &crate::RouteIdentity,
466 ) -> Result<Self, ContextBindingError<Self>> {
467 self.update(|c| {
468 known(
469 &mut c.route,
470 SafeText::metadata(route.as_str()),
471 DiagnosticContextField::Route,
472 )
473 })
474 }
475 pub fn with_module(
476 self,
477 module: &saddle_core::ModuleId,
478 ) -> Result<Self, ContextBindingError<Self>> {
479 self.update(|c| {
480 known(
481 &mut c.module,
482 SafeText::metadata(module.as_str()),
483 DiagnosticContextField::Module,
484 )
485 })
486 }
487 pub fn with_service(
488 self,
489 service: &saddle_core::ServiceId,
490 ) -> Result<Self, ContextBindingError<Self>> {
491 self.update(|c| {
492 known(
493 &mut c.service,
494 SafeText::metadata(service.as_str()),
495 DiagnosticContextField::Service,
496 )
497 })
498 }
499 pub fn with_operation(
500 self,
501 operation: &saddle_core::OperationId,
502 ) -> Result<Self, ContextBindingError<Self>> {
503 self.update(|c| {
504 known(
505 &mut c.operation,
506 SafeText::metadata(operation.as_str()),
507 DiagnosticContextField::Operation,
508 )
509 })
510 }
511}
512
513pub struct RequestDiagnosticScope<'a> {
522 output: Option<&'a EmergencyDiagnosticHandle>,
523 context: Projection,
524}
525#[allow(clippy::result_large_err)]
527impl<'a> RequestDiagnosticScope<'a> {
528 pub fn early(
531 output: Option<&'a EmergencyDiagnosticHandle>,
532 early: EarlyRequestContext,
533 ) -> Self {
534 Self {
535 output,
536 context: early.context,
537 }
538 }
539 pub fn with_output<'b>(
542 self,
543 output: Option<&'b EmergencyDiagnosticHandle>,
544 ) -> RequestDiagnosticScope<'b> {
545 RequestDiagnosticScope {
546 output,
547 context: self.context,
548 }
549 }
550 pub fn reborrow(&self) -> RequestDiagnosticScope<'a> {
554 RequestDiagnosticScope {
555 output: self.output,
556 context: self.context,
557 }
558 }
559 pub fn record_nonfailure(
562 &self,
563 axes: &DiagnosticOutcomeAxes,
564 ) -> Result<DiagnosticSubmission, ()> {
565 if !matches!(
566 axes.operation,
567 saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
568 ) {
569 return Err(());
570 }
571 #[derive(Serialize)]
572 struct NonfailureRecord<'a> {
573 event: &'static str,
574 timestamp_unix_ms: u128,
575 context: &'a Projection,
576 diagnostic: Option<&'a BoundedDiagnostic>,
577 diagnostic_reference: Option<DiagnosticOccurrence>,
578 axes: Option<&'a DiagnosticOutcomeAxes>,
579 source_submission: Option<DiagnosticSubmission>,
580 }
581 Ok(self
582 .output
583 .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
584 output.submit_fixed_record(&NonfailureRecord {
585 event: "framework.boundary.outcome",
586 timestamp_unix_ms: timestamp(),
587 context: &self.context,
588 diagnostic: None::<&BoundedDiagnostic>,
589 diagnostic_reference: None::<DiagnosticOccurrence>,
590 axes: Some(axes),
591 source_submission: None::<DiagnosticSubmission>,
592 })
593 }))
594 }
595 pub fn with_task(mut self, task: DiagnosticTaskId) -> Result<Self, ContextBindingError<Self>> {
596 if let Err(field) = known(
597 &mut self.context.task,
598 SafeText {
599 value: task.0,
600 truncated: false,
601 redacted: false,
602 },
603 DiagnosticContextField::Task,
604 ) {
605 return Err(ContextBindingError {
606 field,
607 original: self,
608 });
609 }
610 Ok(self)
611 }
612 pub fn bind_request_identity(
615 mut self,
616 request: &crate::RequestIdentity,
617 ) -> Result<Self, ContextBindingError<Self>> {
618 if let Err(field) = bind_request(&mut self.context, request.as_str()) {
619 return Err(ContextBindingError {
620 field,
621 original: self,
622 });
623 }
624 Ok(self)
625 }
626
627 pub fn bind_established(
628 mut self,
629 call: &CallContext,
630 event: &EventContext,
631 ) -> Result<Self, ContextBindingError<Self>> {
632 let mut next = self.context;
633 if let Err(field) = bind_call(&mut next, call).and_then(|()| bind_event(&mut next, event)) {
634 return Err(ContextBindingError {
635 field,
636 original: self,
637 });
638 }
639 self.context = next;
640 Ok(self)
641 }
642 pub fn derive_outbound_child(
650 &self,
651 call: &CallContext,
652 event: &EventContext,
653 ) -> Result<RequestDiagnosticScope<'a>, DiagnosticContextField> {
654 use DiagnosticContextField as F;
655 if self.context.trace_id
656 != Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str()))
657 {
658 return Err(F::Trace);
659 }
660 if self.context.request_binding != Some(ProtocolId::copy(event.diagnostic_request())) {
661 return Err(F::Request);
662 }
663 let Field::Present(parent_rpc) = self.context.rpc_id else {
664 return Err(F::Rpc);
665 };
666 let child_rpc = call.rpc_correlation_id().ok_or(F::Rpc)?.as_str();
667 let parent_rpc =
668 std::str::from_utf8(&parent_rpc.bytes[..parent_rpc.len]).map_err(|_| F::Rpc)?;
669 let sequence = child_rpc
670 .strip_prefix(parent_rpc)
671 .and_then(|suffix| suffix.strip_prefix('.'))
672 .ok_or(F::Rpc)?;
673 if sequence.is_empty() || !sequence.bytes().all(|b| b.is_ascii_digit()) {
674 return Err(F::Rpc);
675 }
676 let Field::Present(parent_span) = self.context.span_id else {
677 return Err(F::Span);
678 };
679 if parent_span == Span(call.span_id().as_u64()) {
680 return Err(F::Span);
681 }
682 let child = Projection::existing(call, event);
685 let mut context = self.context;
686 context.application = child.application;
687 context.module = child.module;
688 context.service = child.service;
689 context.operation = child.operation;
690 context.rpc_id = child.rpc_id;
691 context.span_id = child.span_id;
692 context.route = child.route;
693 context.attempt = child.attempt;
694 Ok(RequestDiagnosticScope {
695 output: self.output,
696 context,
697 })
698 }
699 pub fn with_phase(mut self, phase: DiagnosticRequestPhase) -> Self {
701 self.context.lifecycle = Field::Present(SafeText::metadata(phase.as_str()));
702 self
703 }
704 pub fn with_missing(
706 mut self,
707 field: DiagnosticContextField,
708 reason: DiagnosticContextMissing,
709 ) -> Self {
710 fn set<T>(field: &mut Field<T>, reason: DiagnosticContextMissing) {
711 if !matches!(field, Field::Present(_)) {
712 *field = match reason {
713 DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
714 DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
715 DiagnosticContextMissing::Unavailable => Field::Unavailable,
716 };
717 }
718 }
719 match field {
720 DiagnosticContextField::Application => set(&mut self.context.application, reason),
721 DiagnosticContextField::Module => set(&mut self.context.module, reason),
722 DiagnosticContextField::Service => set(&mut self.context.service, reason),
723 DiagnosticContextField::Operation => set(&mut self.context.operation, reason),
724 DiagnosticContextField::DbOperation => set(&mut self.context.db_operation, reason),
725 DiagnosticContextField::Trace => set(&mut self.context.trace_id, reason),
726 DiagnosticContextField::Rpc => set(&mut self.context.rpc_id, reason),
727 DiagnosticContextField::Span => set(&mut self.context.span_id, reason),
728 DiagnosticContextField::Request => set(&mut self.context.request, reason),
729 DiagnosticContextField::Route => set(&mut self.context.route, reason),
730 DiagnosticContextField::Attempt => set(&mut self.context.attempt, reason),
731 DiagnosticContextField::Scope => set(&mut self.context.scope, reason),
732 DiagnosticContextField::Task => set(&mut self.context.task, reason),
733 DiagnosticContextField::Lifecycle => set(&mut self.context.lifecycle, reason),
734 DiagnosticContextField::Zone => set(&mut self.context.zone, reason),
735 DiagnosticContextField::Target => set(&mut self.context.target, reason),
736 }
737 self
738 }
739 pub fn live_db_scope(
741 output: Option<&'a EmergencyDiagnosticHandle>,
742 projection: &saddle_core::DbScopeDiagnosticContext<(&CallContext, &EventContext)>,
743 ) -> Self {
744 let ((call, event), scope) = projection.diagnostic_context();
745 let mut bound = match output {
746 Some(output) => Self::established(output, call, event),
747 None => Self::output_unavailable(call, event),
748 };
749 bound.context.scope = Field::Present(scope);
750 bound
751 }
752
753 pub fn with_db_operation(mut self, operation: DiagnosticDbOperation) -> Self {
755 self.set_db_operation(operation);
756 self
757 }
758 pub fn set_db_operation(&mut self, operation: DiagnosticDbOperation) {
761 self.context.db_operation = Field::Present(operation.0);
762 }
763 pub fn with_db_operation_missing(mut self, reason: DiagnosticContextMissing) -> Self {
764 if !matches!(self.context.db_operation, Field::Present(_)) {
765 self.context.db_operation = match reason {
766 DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
767 DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
768 DiagnosticContextMissing::Unavailable => Field::Unavailable,
769 };
770 }
771 self
772 }
773 pub fn with_zone(mut self, zone: DiagnosticZone) -> Self {
776 self.context.zone = Field::Present(zone.0);
777 self
778 }
779 pub fn with_zone_missing(mut self, reason: DiagnosticContextMissing) -> Self {
780 if !matches!(self.context.zone, Field::Present(_)) {
782 self.context.zone = match reason {
783 DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
784 DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
785 DiagnosticContextMissing::Unavailable => Field::Unavailable,
786 };
787 }
788 self
789 }
790 pub fn db_scope(
793 output: &'a EmergencyDiagnosticHandle,
794 observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
795 ) -> Self {
796 let ((_, call, event), scope) = observation.diagnostic_context();
797 let mut bound = Self::established(output, call, event);
798 bound.context.scope = Field::Present(scope);
799 bound
800 }
801 pub fn db_scope_output_unavailable(
802 observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
803 ) -> Self {
804 let ((_, call, event), scope) = observation.diagnostic_context();
805 let mut bound = Self::output_unavailable(call, event);
806 bound.context.scope = Field::Present(scope);
807 bound
808 }
809 pub fn established(
810 output: &'a EmergencyDiagnosticHandle,
811 call: &CallContext,
812 event: &EventContext,
813 ) -> Self {
814 Self {
815 output: Some(output),
816 context: Projection::existing(call, event),
817 }
818 }
819
820 pub fn output_unavailable(call: &CallContext, event: &EventContext) -> Self {
823 Self {
824 output: None,
825 context: Projection::existing(call, event),
826 }
827 }
828 pub fn capture_required(&self, diagnostic: BoundedDiagnostic) -> RequestSourceReceipt {
830 self.capture_error((), diagnostic)
831 }
832 pub fn capture_existing(
837 &self,
838 diagnostic: saddle_core::Diagnostic,
839 observer: Option<&crate::Observer>,
840 ) -> ExistingDiagnosticReceipt {
841 let occurrence = diagnostic.occurrence();
842 let record = Record {
843 event: "framework.diagnostic",
844 timestamp_unix_ms: timestamp(),
845 context: &self.context,
846 diagnostic: Some(&diagnostic),
847 diagnostic_reference: occurrence,
848 axes: None,
849 source_submission: None,
850 };
851 let submission = self
852 .output
853 .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
854 output.submit_existing_record(&record, diagnostic.deferred_stack())
855 });
856 if let Some(observer) = observer {
857 observer.mirror_existing_record(&record, diagnostic.category());
858 }
859 FrameworkRequestFailure {
860 error: (),
861 context: self.context,
862 occurrence,
863 submission,
864 diagnostic,
865 }
866 }
867
868 #[track_caller]
871 pub fn fail<E>(
872 &self,
873 error: E,
874 category: DiagnosticCategory,
875 cause: BoundedDiagnosticCause,
876 ) -> FrameworkRequestFailure<E> {
877 let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
878 self.capture_error(error, diagnostic)
879 }
880 fn capture_error<E>(
881 &self,
882 error: E,
883 diagnostic: BoundedDiagnostic,
884 ) -> FrameworkRequestFailure<E> {
885 let occurrence = diagnostic.occurrence();
886 let submission = self
887 .output
888 .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
889 output.submit_fixed_record(&Record {
890 event: "framework.diagnostic",
891 timestamp_unix_ms: timestamp(),
892 context: &self.context,
893 diagnostic: Some(&diagnostic),
894 diagnostic_reference: occurrence,
895 axes: None,
896 source_submission: None,
897 })
898 });
899 FrameworkRequestFailure {
900 error,
901 context: self.context,
902 occurrence,
903 submission,
904 diagnostic,
905 }
906 }
907}
908
909#[must_use = "retain the source receipt with the technical result until its declared boundary"]
931pub struct FrameworkRequestFailure<E, D = BoundedDiagnostic> {
932 error: E,
933 context: Projection,
934 occurrence: DiagnosticOccurrence,
935 submission: DiagnosticSubmission,
936 diagnostic: D,
937}
938pub type RequestSourceReceipt = FrameworkRequestFailure<()>;
950pub type ExistingDiagnosticReceipt = FrameworkRequestFailure<(), saddle_core::Diagnostic>;
951impl<E, D> FrameworkRequestFailure<E, D> {
952 pub fn error(&self) -> &E {
953 &self.error
954 }
955 pub fn submission(&self) -> DiagnosticSubmission {
956 self.submission
957 }
958 pub fn map_error<F>(self, map: impl FnOnce(E) -> F) -> FrameworkRequestFailure<F, D> {
959 FrameworkRequestFailure {
960 error: map(self.error),
961 context: self.context,
962 occurrence: self.occurrence,
963 submission: self.submission,
964 diagnostic: self.diagnostic,
965 }
966 }
967 pub fn record_boundary(
968 &self,
969 output: &EmergencyDiagnosticHandle,
970 axes: &DiagnosticOutcomeAxes,
971 ) -> DiagnosticSubmission {
972 self.record_boundary_optional(Some(output), axes)
973 }
974 pub fn record_boundary_optional(
975 &self,
976 output: Option<&EmergencyDiagnosticHandle>,
977 axes: &DiagnosticOutcomeAxes,
978 ) -> DiagnosticSubmission {
979 output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
980 output.submit_fixed_record(&Record {
981 event: "framework.boundary.outcome",
982 timestamp_unix_ms: timestamp(),
983 context: &self.context,
984 diagnostic: None::<&BoundedDiagnostic>,
985 diagnostic_reference: self.occurrence,
986 axes: Some(axes),
987 source_submission: Some(self.submission),
988 })
989 })
990 }
991 pub fn finish_boundary_retained(
996 self,
997 output: Option<&EmergencyDiagnosticHandle>,
998 axes: &DiagnosticOutcomeAxes,
999 ) -> (Self, BoundaryDiagnosticDelivery) {
1000 let delivery = BoundaryDiagnosticDelivery {
1001 source: self.submission,
1002 boundary: self.record_boundary_optional(output, axes),
1003 occurrence: self.occurrence,
1004 };
1005 (self, delivery)
1006 }
1007 pub fn source_diagnostic(&self) -> &D {
1008 &self.diagnostic
1009 }
1010 pub fn finish_boundary(
1014 self,
1015 output: &EmergencyDiagnosticHandle,
1016 axes: &DiagnosticOutcomeAxes,
1017 ) -> (E, BoundaryDiagnosticDelivery) {
1018 let boundary = self.record_boundary(output, axes);
1019 (
1020 self.error,
1021 BoundaryDiagnosticDelivery {
1022 source: self.submission,
1023 boundary,
1024 occurrence: self.occurrence,
1025 },
1026 )
1027 }
1028}
1029impl<D> FrameworkRequestFailure<(), D> {
1030 pub fn into_reference(self) -> RequestBoundaryReference<D> {
1032 RequestBoundaryReference {
1033 context: self.context,
1034 occurrence: self.occurrence,
1035 submission: self.submission,
1036 diagnostic: self.diagnostic,
1037 }
1038 }
1039}
1040#[must_use = "retain source facts and submission status through terminal consumption"]
1041pub struct RequestBoundaryReference<D = BoundedDiagnostic> {
1042 context: Projection,
1043 occurrence: DiagnosticOccurrence,
1044 submission: DiagnosticSubmission,
1045 diagnostic: D,
1046}
1047impl<D> RequestBoundaryReference<D> {
1048 pub(crate) fn record_stage(
1049 &self,
1050 output: Option<&EmergencyDiagnosticHandle>,
1051 axes: &DiagnosticOutcomeAxes,
1052 stage: &'static str,
1053 elapsed_ms: u64,
1054 ) -> BoundaryDiagnosticDelivery {
1055 #[derive(Serialize)]
1056 struct StageRecord<'a> {
1057 #[serde(flatten)]
1058 record: Record<'a>,
1059 stage: &'static str,
1060 elapsed_ms: u64,
1061 }
1062 let boundary = output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
1063 output.submit_fixed_record(&StageRecord {
1064 record: Record {
1065 event: "framework.boundary.outcome",
1066 timestamp_unix_ms: timestamp(),
1067 context: &self.context,
1068 diagnostic: None,
1069 diagnostic_reference: self.occurrence,
1070 axes: Some(axes),
1071 source_submission: Some(self.submission),
1072 },
1073 stage,
1074 elapsed_ms,
1075 })
1076 });
1077 BoundaryDiagnosticDelivery {
1078 source: self.submission,
1079 boundary,
1080 occurrence: self.occurrence,
1081 }
1082 }
1083 pub fn occurrence(&self) -> DiagnosticOccurrence {
1084 self.occurrence
1085 }
1086 pub fn source_submission(&self) -> DiagnosticSubmission {
1087 self.submission
1088 }
1089 pub fn source_diagnostic(&self) -> &D {
1091 &self.diagnostic
1092 }
1093 pub fn record(
1094 &self,
1095 output: &EmergencyDiagnosticHandle,
1096 axes: &DiagnosticOutcomeAxes,
1097 ) -> DiagnosticSubmission {
1098 self.record_optional(Some(output), axes)
1099 }
1100 pub fn record_optional(
1101 &self,
1102 output: Option<&EmergencyDiagnosticHandle>,
1103 axes: &DiagnosticOutcomeAxes,
1104 ) -> DiagnosticSubmission {
1105 output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
1106 output.submit_fixed_record(&Record {
1107 event: "framework.boundary.outcome",
1108 timestamp_unix_ms: timestamp(),
1109 context: &self.context,
1110 diagnostic: None::<&BoundedDiagnostic>,
1111 diagnostic_reference: self.occurrence,
1112 axes: Some(axes),
1113 source_submission: Some(self.submission),
1114 })
1115 })
1116 }
1117 pub fn finish_retained(
1119 self,
1120 output: Option<&EmergencyDiagnosticHandle>,
1121 axes: &DiagnosticOutcomeAxes,
1122 ) -> (Self, BoundaryDiagnosticDelivery) {
1123 let delivery = BoundaryDiagnosticDelivery {
1124 source: self.submission,
1125 boundary: self.record_optional(output, axes),
1126 occurrence: self.occurrence,
1127 };
1128 (self, delivery)
1129 }
1130}
1131pub struct BoundaryDiagnosticDelivery {
1133 source: DiagnosticSubmission,
1134 boundary: DiagnosticSubmission,
1135 occurrence: DiagnosticOccurrence,
1136}
1137impl BoundaryDiagnosticDelivery {
1138 pub fn source_submission(&self) -> DiagnosticSubmission {
1139 self.source
1140 }
1141 pub fn boundary_submission(&self) -> DiagnosticSubmission {
1142 self.boundary
1143 }
1144 pub fn occurrence(&self) -> DiagnosticOccurrence {
1145 self.occurrence
1146 }
1147}
1148#[derive(Serialize)]
1149struct Record<'a, D = BoundedDiagnostic> {
1150 event: &'static str,
1151 timestamp_unix_ms: u128,
1152 context: &'a Projection,
1153 diagnostic: Option<&'a D>,
1154 diagnostic_reference: DiagnosticOccurrence,
1155 axes: Option<&'a DiagnosticOutcomeAxes>,
1156 #[serde(skip_serializing_if = "Option::is_none")]
1157 source_submission: Option<DiagnosticSubmission>,
1158}
1159fn timestamp() -> u128 {
1160 std::time::SystemTime::now()
1161 .duration_since(std::time::UNIX_EPOCH)
1162 .unwrap_or_default()
1163 .as_millis()
1164}
1165
1166#[cfg(test)]
1167mod early_tests {
1168 use super::*;
1169 use saddle_core::*;
1170 fn ok<T>(value: std::result::Result<T, ContextBindingError<T>>) -> T {
1171 match value {
1172 Ok(value) => value,
1173 Err(_) => panic!("unexpected context conflict"),
1174 }
1175 }
1176 fn cause() -> BoundedDiagnosticCause {
1177 BoundedDiagnosticCause::new(
1178 DiagnosticStage::RequestDecode,
1179 DiagnosticCode::new("request.read_failed").unwrap(),
1180 )
1181 }
1182 fn context(scope: &RequestDiagnosticScope<'_>) -> serde_json::Value {
1183 serde_json::to_value(scope.context).unwrap()
1184 }
1185 #[test]
1186 fn early_absence_and_output_are_independent_and_receipt_retains_error() {
1187 let scope =
1188 RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("app"))
1189 .with_missing(
1190 DiagnosticContextField::DbOperation,
1191 DiagnosticContextMissing::NotApplicable,
1192 )
1193 .with_missing(
1194 DiagnosticContextField::Application,
1195 DiagnosticContextMissing::Unavailable,
1196 )
1197 .with_phase(DiagnosticRequestPhase::ReadingHead);
1198 let value = context(&scope);
1199 assert_eq!(value["application"]["value"]["value"], "app");
1200 for field in ["trace_id", "rpc_id", "span_id", "request", "scope", "task"] {
1201 assert_eq!(value[field]["state"], "not_established");
1202 assert!(value[field].get("value").is_none());
1203 }
1204 assert_eq!(value["db_operation"]["state"], "not_applicable");
1205 assert_eq!(value["lifecycle"]["value"]["value"], "reading_head");
1206 let failure = scope.fail(17u32, DiagnosticCategory::UnexpectedError, cause());
1207 let id = failure.source_diagnostic().id();
1208 assert_eq!(
1209 failure.submission(),
1210 DiagnosticSubmission::OutputUnavailable
1211 );
1212 let mapped = failure.map_error(|code| code + 1);
1213 let (retained, delivery) =
1214 mapped.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
1215 assert_eq!(*retained.error(), 18);
1216 assert_eq!(retained.source_diagnostic().id(), id);
1217 assert_eq!(
1218 delivery.source_submission(),
1219 DiagnosticSubmission::OutputUnavailable
1220 );
1221 assert_eq!(
1222 delivery.boundary_submission(),
1223 DiagnosticSubmission::OutputUnavailable
1224 );
1225 let lost = RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable());
1226 assert_eq!(context(&lost)["trace_id"]["state"], "unavailable");
1227 }
1228 #[test]
1229 fn known_partial_identity_atomic_conflict_and_promotion_preserve_snapshots() {
1230 let trace = TraceCorrelationId::new("gateway-opaque-原值").unwrap();
1231 let foreign = TraceCorrelationId::new("foreign").unwrap();
1232 let early = ok(EarlyRequestContext::socket_accepted("app").with_trace(&trace));
1233 let early = match early.with_trace(&foreign) {
1234 Ok(_) => panic!("foreign trace accepted"),
1235 Err(error) => {
1236 assert_eq!(error.field(), DiagnosticContextField::Trace);
1237 error.into_original()
1238 }
1239 };
1240 let request = crate::RequestIdentity::new("request-1").unwrap();
1241 let early = ok(early.with_request(&request));
1242 let scope = ok(RequestDiagnosticScope::early(None, early)
1243 .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
1244 let before = context(&scope);
1245 let receipt = scope
1246 .fail((), DiagnosticCategory::UnexpectedError, cause())
1247 .into_reference();
1248 let call = CallContext::new(
1249 "app".into(),
1250 "module".into(),
1251 "service".into(),
1252 "operation".into(),
1253 TraceId::from_u128(1),
1254 SpanId::from_u64(2),
1255 )
1256 .with_trace_correlation_id(foreign);
1257 let event =
1258 EventContext::new(request, crate::RouteIdentity::new("/route").unwrap(), 1).unwrap();
1259 let scope = match scope.bind_established(&call, &event) {
1260 Ok(_) => panic!("foreign promotion accepted"),
1261 Err(error) => error.into_original(),
1262 };
1263 assert_eq!(context(&scope), before); let call = call.with_trace_correlation_id(trace);
1265 let scope = ok(scope.bind_established(&call, &event));
1266 let after = context(&scope);
1267 assert_eq!(after["trace_id"]["value"], "gateway-opaque-原值");
1268 assert_eq!(after["task"]["value"]["value"], "42");
1269 assert_eq!(after["span_id"]["value"], "0000000000000002");
1270 assert_eq!(serde_json::to_value(receipt.context).unwrap(), before);
1271 assert!(
1272 scope.with_output(None).context.request
1273 == Field::Present(SafeText::metadata("request-1"))
1274 );
1275 }
1276 #[test]
1277 fn task_projection_is_bounded_and_cannot_be_replaced() {
1278 for unsafe_id in ["", "request_task", "-1", "https://secret", "12\n3"] {
1279 assert!(DiagnosticTaskId::from_runtime_id(unsafe_id).is_none());
1280 }
1281 assert!(DiagnosticTaskId::from_runtime_id(&"1".repeat(65)).is_none());
1282 let scope = ok(RequestDiagnosticScope::early(
1283 None,
1284 EarlyRequestContext::socket_accepted("app"),
1285 )
1286 .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
1287 let original = context(&scope);
1288 let scope = match scope.with_task(DiagnosticTaskId::from_runtime_id("43").unwrap()) {
1289 Ok(_) => panic!("changed task identity"),
1290 Err(error) => error.into_original(),
1291 };
1292 assert_eq!(context(&scope), original);
1293 }
1294}
1295
1296#[cfg(test)]
1297mod outbound_child_tests {
1298 use super::*;
1299 use saddle_core::*;
1300
1301 fn event(request: &str, route: &str) -> EventContext {
1302 EventContext::new(
1303 crate::RequestIdentity::new(request).unwrap(),
1304 crate::RouteIdentity::new(route).unwrap(),
1305 1,
1306 )
1307 .unwrap()
1308 }
1309 fn call(trace: &str, rpc: &str, span: u64) -> CallContext {
1310 CallContext::new(
1311 "saddle".into(),
1312 "zone-a".into(),
1313 "profusecontract".into(),
1314 "invoke".into(),
1315 TraceId::from_u128(1),
1316 SpanId::from_u64(span),
1317 )
1318 .with_trace_correlation_id(TraceCorrelationId::new(trace).unwrap())
1319 .with_rpc_correlation_id(RpcCorrelationId::new(rpc))
1320 }
1321 fn cause() -> BoundedDiagnosticCause {
1322 BoundedDiagnosticCause::new(
1323 DiagnosticStage::RequestDecode,
1324 DiagnosticCode::new("outbound.connect_failed").unwrap(),
1325 )
1326 }
1327 #[test]
1328 fn outbound_child_preserves_live_scope_and_receipts() {
1329 let parent = call("opaque-trace", "0.4", 1);
1330 let child = call("opaque-trace", "0.4.1", 2);
1331 let parent_event = event("request-1", "/incoming");
1332 let child_event = event("request-1", "remote.function");
1333 let (_, issuer) = DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
1334 let (request, execution) = issuer.issue_request().unwrap();
1335 let live = request
1336 .project_diagnostic_context(&execution, (&parent, &parent_event))
1337 .ok()
1338 .unwrap();
1339 let scope = RequestDiagnosticScope::live_db_scope(None, &live)
1340 .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
1341 .unwrap_or_else(|_| panic!("task"))
1342 .with_zone(DiagnosticZone::from_validated_ingress("zone-a").unwrap())
1343 .with_db_operation(DiagnosticDbOperation::from_registered("orders.query").unwrap());
1344 let original = serde_json::to_value(scope.context).unwrap();
1345 let old = scope.fail(7, DiagnosticCategory::UnexpectedError, cause());
1346 let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
1347 let value = serde_json::to_value(derived.context).unwrap();
1348 for field in [
1349 "request",
1350 "trace_id",
1351 "task",
1352 "zone",
1353 "scope",
1354 "db_operation",
1355 "lifecycle",
1356 "target",
1357 ] {
1358 assert_eq!(value[field], original[field], "{field}");
1359 }
1360 assert_eq!(value["rpc_id"]["value"], "0.4.1");
1361 assert_eq!(value["span_id"]["value"], "0000000000000002");
1362 assert_eq!(value["route"]["value"]["value"], "remote.function");
1363 assert_eq!(serde_json::to_value(scope.context).unwrap(), original);
1364 assert_eq!(serde_json::to_value(old.context).unwrap(), original);
1365 assert!(
1366 scope
1367 .reborrow()
1368 .bind_established(&child, &child_event)
1369 .is_err()
1370 );
1371 let source = derived.fail(9, DiagnosticCategory::UnexpectedError, cause());
1372 let id = source.source_diagnostic().id();
1373 let (retained, delivery) =
1374 source.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
1375 assert_eq!(retained.source_diagnostic().id(), id);
1376 assert_eq!(*retained.error(), 9);
1377 assert_eq!(
1378 delivery.source_submission(),
1379 DiagnosticSubmission::OutputUnavailable
1380 );
1381 assert_eq!(
1382 delivery.boundary_submission(),
1383 DiagnosticSubmission::OutputUnavailable
1384 );
1385 assert_eq!(serde_json::to_value(retained.context).unwrap(), value);
1386 }
1387
1388 #[test]
1389 fn outbound_child_rejects_foreign_missing_and_redaction_collisions() {
1390 let parent = call("trace", "0", 1);
1391 let scope = RequestDiagnosticScope::output_unavailable(&parent, &event("a@b", "/in"));
1393 let before = serde_json::to_value(scope.context).unwrap();
1394 let child_event = event("a@b", "remote");
1395 for (child, ev, expected) in [
1396 (
1397 call("foreign", "0.1", 2),
1398 child_event.clone(),
1399 DiagnosticContextField::Trace,
1400 ),
1401 (
1402 call("trace", "0.1", 2),
1403 event("c@d", "remote"),
1404 DiagnosticContextField::Request,
1405 ),
1406 (
1407 call("trace", "01.1", 2),
1408 child_event.clone(),
1409 DiagnosticContextField::Rpc,
1410 ),
1411 (
1412 call("trace", "0", 2),
1413 child_event.clone(),
1414 DiagnosticContextField::Rpc,
1415 ),
1416 (
1417 call("trace", "0.1.2", 2),
1418 child_event.clone(),
1419 DiagnosticContextField::Rpc,
1420 ),
1421 (
1422 call("trace", "0.x", 2),
1423 child_event.clone(),
1424 DiagnosticContextField::Rpc,
1425 ),
1426 (
1427 call("trace", "0.1", 1),
1428 child_event.clone(),
1429 DiagnosticContextField::Span,
1430 ),
1431 ] {
1432 assert_eq!(
1433 scope.derive_outbound_child(&child, &ev).err(),
1434 Some(expected)
1435 );
1436 assert_eq!(serde_json::to_value(scope.context).unwrap(), before);
1437 }
1438 let child = call("trace", "0.1", 2);
1439 let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
1440 let json = serde_json::to_string(&derived.context).unwrap();
1441 assert!(!json.contains("a@b") && !json.contains("request_binding"));
1442 assert!(
1443 RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable())
1444 .derive_outbound_child(&child, &child_event)
1445 .is_err()
1446 );
1447 let missing_rpc = parent.clone().with_rpc_correlation_id(None);
1448 assert_eq!(
1449 RequestDiagnosticScope::output_unavailable(&missing_rpc, &child_event)
1450 .derive_outbound_child(&child, &child_event)
1451 .err(),
1452 Some(DiagnosticContextField::Rpc)
1453 );
1454 assert!(
1455 scope
1456 .reborrow()
1457 .bind_request_identity(&crate::RequestIdentity::new("c@d").unwrap())
1458 .is_err()
1459 );
1460 assert!(
1461 EarlyRequestContext::unavailable()
1462 .with_request(&crate::RequestIdentity::new("a@b").unwrap())
1463 .unwrap_or_else(|_| panic!("first identity"))
1464 .with_request(&crate::RequestIdentity::new("c@d").unwrap())
1465 .is_err()
1466 );
1467 }
1468}