use crate::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext};
use saddle_core::{
BoundedDiagnostic, BoundedDiagnosticCause, CallContext, CaptureSite, DiagnosticCategory,
DiagnosticOccurrence, DiagnosticOutcomeAxes,
};
use serde::{Serialize, Serializer};
#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
#[serde(tag = "state", content = "value", rename_all = "snake_case")]
enum Field<T> {
Present(T),
NotApplicable,
NotEstablished,
Unavailable,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DiagnosticContextMissing {
NotApplicable,
NotEstablished,
Unavailable,
}
#[derive(Clone, Copy)]
pub struct DiagnosticDbOperation(&'static str);
impl DiagnosticDbOperation {
pub fn from_registered(value: &'static str) -> Option<Self> {
(!value.is_empty()
&& value.len() <= 128
&& value
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b"._:-".contains(&b)))
.then_some(Self(value))
}
}
#[derive(Clone, Copy)]
pub struct DiagnosticZone(ProtocolId);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DiagnosticZoneError {
Empty,
TooLong,
Unsafe,
}
impl DiagnosticZone {
pub fn from_validated_ingress(value: &str) -> Result<Self, DiagnosticZoneError> {
if value.len() > 256 {
return Err(DiagnosticZoneError::TooLong);
}
if value.trim().is_empty() {
return Err(DiagnosticZoneError::Empty);
}
if value.chars().any(char::is_control)
|| value.contains("://")
|| value.contains(['@', '?', '#', '\\'])
{
return Err(DiagnosticZoneError::Unsafe);
}
Ok(Self(ProtocolId::copy(value)))
}
}
#[derive(Clone, Copy, Eq, PartialEq)]
struct SafeText {
value: ProtocolId,
truncated: bool,
redacted: bool,
}
impl SafeText {
fn metadata(value: &str) -> Self {
let mut len = value.len().min(256);
while !value.is_char_boundary(len) {
len -= 1;
}
let prefix = &value[..len];
let redacted = prefix.contains("://")
|| prefix
.chars()
.any(|c| !(c.is_alphanumeric() || "_./:{}*-".contains(c)));
Self {
value: ProtocolId::copy(if redacted { "" } else { prefix }),
truncated: len < value.len(),
redacted,
}
}
}
impl Serialize for SafeText {
fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
use serde::ser::SerializeStruct;
let mut out = s.serialize_struct("SafeText", 3)?;
out.serialize_field("value", &self.value)?;
out.serialize_field("truncated", &self.truncated)?;
out.serialize_field("redacted", &self.redacted)?;
out.end()
}
}
#[derive(Clone, Copy, Eq, PartialEq)]
struct Span(u64);
impl Serialize for Span {
fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
let mut bytes = [b'0'; 16];
for (i, b) in bytes.iter_mut().enumerate() {
*b = b"0123456789abcdef"[((self.0 >> ((15 - i) * 4)) & 15) as usize];
}
s.serialize_str(std::str::from_utf8(&bytes).unwrap())
}
}
#[derive(Clone, Copy, Eq, PartialEq)]
struct ProtocolId {
bytes: [u8; 256],
len: usize,
}
impl ProtocolId {
fn copy(value: &str) -> Self {
let mut bytes = [0; 256];
bytes[..value.len()].copy_from_slice(value.as_bytes());
Self {
bytes,
len: value.len(),
}
}
}
impl Serialize for ProtocolId {
fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
s.serialize_str(std::str::from_utf8(&self.bytes[..self.len]).unwrap_or(""))
}
}
#[derive(Clone, Copy, Serialize)]
struct Projection {
schema_version: u8,
application: Field<SafeText>,
module: Field<SafeText>,
service: Field<SafeText>,
operation: Field<SafeText>,
db_operation: Field<&'static str>,
trace_id: Field<ProtocolId>,
rpc_id: Field<ProtocolId>,
span_id: Field<Span>,
request: Field<SafeText>,
#[serde(skip)]
request_binding: Option<ProtocolId>,
route: Field<SafeText>,
attempt: Field<u32>,
scope: Field<saddle_core::DbScopeDiagnosticIdentity>,
task: Field<SafeText>,
lifecycle: Field<SafeText>,
zone: Field<ProtocolId>,
target: Field<SafeText>,
}
impl Projection {
fn missing(reason: DiagnosticContextMissing) -> Self {
fn absent<T>(reason: DiagnosticContextMissing) -> Field<T> {
match reason {
DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
DiagnosticContextMissing::Unavailable => Field::Unavailable,
}
}
Self {
schema_version: 1,
application: absent(reason),
module: absent(reason),
service: absent(reason),
operation: absent(reason),
db_operation: absent(reason),
trace_id: absent(reason),
rpc_id: absent(reason),
span_id: absent(reason),
request: absent(reason),
request_binding: None,
route: absent(reason),
attempt: absent(reason),
scope: absent(reason),
task: absent(reason),
lifecycle: absent(reason),
zone: absent(reason),
target: absent(reason),
}
}
fn existing(call: &CallContext, event: &EventContext) -> Self {
let text = |s: &str| Field::Present(SafeText::metadata(s));
Self {
schema_version: 1,
application: text(call.application().as_str()),
module: text(call.module().as_str()),
service: text(call.service().as_str()),
operation: text(call.operation().as_str()),
db_operation: Field::Unavailable,
trace_id: Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str())),
rpc_id: call.rpc_correlation_id().map_or(Field::Unavailable, |id| {
Field::Present(ProtocolId::copy(id.as_str()))
}),
span_id: Field::Present(Span(call.span_id().as_u64())),
request: text(event.diagnostic_request()),
request_binding: Some(ProtocolId::copy(event.diagnostic_request())),
route: text(event.diagnostic_route()),
attempt: Field::Present(event.diagnostic_attempt()),
scope: Field::Unavailable,
task: Field::Unavailable,
lifecycle: Field::Unavailable,
zone: Field::Unavailable,
target: Field::Unavailable,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DiagnosticContextField {
Application,
Module,
Service,
Operation,
DbOperation,
Trace,
Rpc,
Span,
Request,
Route,
Attempt,
Scope,
Task,
Lifecycle,
Zone,
Target,
}
#[must_use = "recover the original context; do not silently discard known identity"]
pub struct ContextBindingError<T> {
field: DiagnosticContextField,
original: T,
}
impl<T> ContextBindingError<T> {
pub fn field(&self) -> DiagnosticContextField {
self.field
}
pub fn into_original(self) -> T {
self.original
}
}
fn known<T: Copy + Eq>(
field: &mut Field<T>,
value: T,
name: DiagnosticContextField,
) -> Result<(), DiagnosticContextField> {
if let Field::Present(old) = field
&& *old != value
{
return Err(name);
}
*field = Field::Present(value);
Ok(())
}
fn bind_call(context: &mut Projection, call: &CallContext) -> Result<(), DiagnosticContextField> {
use DiagnosticContextField as F;
known(
&mut context.application,
SafeText::metadata(call.application().as_str()),
F::Application,
)?;
known(
&mut context.module,
SafeText::metadata(call.module().as_str()),
F::Module,
)?;
known(
&mut context.service,
SafeText::metadata(call.service().as_str()),
F::Service,
)?;
known(
&mut context.operation,
SafeText::metadata(call.operation().as_str()),
F::Operation,
)?;
known(
&mut context.trace_id,
ProtocolId::copy(call.trace_correlation_id().as_str()),
F::Trace,
)?;
known(&mut context.span_id, Span(call.span_id().as_u64()), F::Span)?;
if let Some(rpc) = call.rpc_correlation_id() {
known(&mut context.rpc_id, ProtocolId::copy(rpc.as_str()), F::Rpc)?;
}
Ok(())
}
fn bind_event(
context: &mut Projection,
event: &EventContext,
) -> Result<(), DiagnosticContextField> {
use DiagnosticContextField as F;
bind_request(context, event.diagnostic_request())?;
known(
&mut context.route,
SafeText::metadata(event.diagnostic_route()),
F::Route,
)?;
known(&mut context.attempt, event.diagnostic_attempt(), F::Attempt)
}
fn bind_request(context: &mut Projection, request: &str) -> Result<(), DiagnosticContextField> {
let exact = ProtocolId::copy(request);
if context.request_binding.is_some_and(|old| old != exact) {
return Err(DiagnosticContextField::Request);
}
known(
&mut context.request,
SafeText::metadata(request),
DiagnosticContextField::Request,
)?;
context.request_binding = Some(exact);
Ok(())
}
#[derive(Clone, Copy)]
pub enum DiagnosticRequestPhase {
SocketAccepted,
Admission,
ReadingHead,
ReadingBody,
Validation,
Dispatch,
Response,
TaskJoin,
Finalization,
}
impl DiagnosticRequestPhase {
fn as_str(self) -> &'static str {
match self {
Self::SocketAccepted => "socket_accepted",
Self::Admission => "admission",
Self::ReadingHead => "reading_head",
Self::ReadingBody => "reading_body",
Self::Validation => "validation",
Self::Dispatch => "dispatch",
Self::Response => "response",
Self::TaskJoin => "task_join",
Self::Finalization => "finalization",
}
}
}
#[derive(Clone, Copy)]
pub struct DiagnosticTaskId(ProtocolId);
impl DiagnosticTaskId {
pub fn from_runtime_id(value: &str) -> Option<Self> {
(!value.is_empty() && value.len() <= 64 && value.bytes().all(|b| b.is_ascii_digit()))
.then(|| Self(ProtocolId::copy(value)))
}
}
pub struct EarlyRequestContext {
context: Projection,
}
#[allow(clippy::result_large_err)]
impl EarlyRequestContext {
pub fn socket_accepted(application: &str) -> Self {
let mut context = Projection::missing(DiagnosticContextMissing::NotEstablished);
context.application = Field::Present(SafeText::metadata(application));
context.lifecycle = Field::Present(SafeText::metadata("socket_accepted"));
Self { context }
}
pub fn unavailable() -> Self {
Self {
context: Projection::missing(DiagnosticContextMissing::Unavailable),
}
}
fn update(
mut self,
bind: impl FnOnce(&mut Projection) -> Result<(), DiagnosticContextField>,
) -> Result<Self, ContextBindingError<Self>> {
let mut next = self.context;
if let Err(field) = bind(&mut next) {
return Err(ContextBindingError {
field,
original: self,
});
}
self.context = next;
Ok(self)
}
pub fn with_application(
self,
application: &saddle_core::ApplicationId,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.application,
SafeText::metadata(application.as_str()),
DiagnosticContextField::Application,
)
})
}
pub fn with_trace(
self,
trace: &saddle_core::TraceCorrelationId,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.trace_id,
ProtocolId::copy(trace.as_str()),
DiagnosticContextField::Trace,
)
})
}
pub fn with_rpc(
self,
rpc: &saddle_core::RpcCorrelationId,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.rpc_id,
ProtocolId::copy(rpc.as_str()),
DiagnosticContextField::Rpc,
)
})
}
pub fn with_event(self, event: &EventContext) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| bind_event(c, event))
}
pub fn with_call(self, call: &CallContext) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| bind_call(c, call))
}
pub fn with_request(
self,
request: &crate::RequestIdentity,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| bind_request(c, request.as_str()))
}
pub fn with_route(
self,
route: &crate::RouteIdentity,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.route,
SafeText::metadata(route.as_str()),
DiagnosticContextField::Route,
)
})
}
pub fn with_module(
self,
module: &saddle_core::ModuleId,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.module,
SafeText::metadata(module.as_str()),
DiagnosticContextField::Module,
)
})
}
pub fn with_service(
self,
service: &saddle_core::ServiceId,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.service,
SafeText::metadata(service.as_str()),
DiagnosticContextField::Service,
)
})
}
pub fn with_operation(
self,
operation: &saddle_core::OperationId,
) -> Result<Self, ContextBindingError<Self>> {
self.update(|c| {
known(
&mut c.operation,
SafeText::metadata(operation.as_str()),
DiagnosticContextField::Operation,
)
})
}
}
pub struct RequestDiagnosticScope<'a> {
output: Option<&'a EmergencyDiagnosticHandle>,
context: Projection,
}
#[allow(clippy::result_large_err)]
impl<'a> RequestDiagnosticScope<'a> {
pub fn early(
output: Option<&'a EmergencyDiagnosticHandle>,
early: EarlyRequestContext,
) -> Self {
Self {
output,
context: early.context,
}
}
pub fn with_output<'b>(
self,
output: Option<&'b EmergencyDiagnosticHandle>,
) -> RequestDiagnosticScope<'b> {
RequestDiagnosticScope {
output,
context: self.context,
}
}
pub fn reborrow(&self) -> RequestDiagnosticScope<'a> {
RequestDiagnosticScope {
output: self.output,
context: self.context,
}
}
pub fn record_nonfailure(
&self,
axes: &DiagnosticOutcomeAxes,
) -> Result<DiagnosticSubmission, ()> {
if !matches!(
axes.operation,
saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
) {
return Err(());
}
#[derive(Serialize)]
struct NonfailureRecord<'a> {
event: &'static str,
timestamp_unix_ms: u128,
context: &'a Projection,
diagnostic: Option<&'a BoundedDiagnostic>,
diagnostic_reference: Option<DiagnosticOccurrence>,
axes: Option<&'a DiagnosticOutcomeAxes>,
source_submission: Option<DiagnosticSubmission>,
}
Ok(self
.output
.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
output.submit_fixed_record(&NonfailureRecord {
event: "framework.boundary.outcome",
timestamp_unix_ms: timestamp(),
context: &self.context,
diagnostic: None::<&BoundedDiagnostic>,
diagnostic_reference: None::<DiagnosticOccurrence>,
axes: Some(axes),
source_submission: None::<DiagnosticSubmission>,
})
}))
}
pub fn with_task(mut self, task: DiagnosticTaskId) -> Result<Self, ContextBindingError<Self>> {
if let Err(field) = known(
&mut self.context.task,
SafeText {
value: task.0,
truncated: false,
redacted: false,
},
DiagnosticContextField::Task,
) {
return Err(ContextBindingError {
field,
original: self,
});
}
Ok(self)
}
pub fn bind_request_identity(
mut self,
request: &crate::RequestIdentity,
) -> Result<Self, ContextBindingError<Self>> {
if let Err(field) = bind_request(&mut self.context, request.as_str()) {
return Err(ContextBindingError {
field,
original: self,
});
}
Ok(self)
}
pub fn bind_established(
mut self,
call: &CallContext,
event: &EventContext,
) -> Result<Self, ContextBindingError<Self>> {
let mut next = self.context;
if let Err(field) = bind_call(&mut next, call).and_then(|()| bind_event(&mut next, event)) {
return Err(ContextBindingError {
field,
original: self,
});
}
self.context = next;
Ok(self)
}
pub fn derive_outbound_child(
&self,
call: &CallContext,
event: &EventContext,
) -> Result<RequestDiagnosticScope<'a>, DiagnosticContextField> {
use DiagnosticContextField as F;
if self.context.trace_id
!= Field::Present(ProtocolId::copy(call.trace_correlation_id().as_str()))
{
return Err(F::Trace);
}
if self.context.request_binding != Some(ProtocolId::copy(event.diagnostic_request())) {
return Err(F::Request);
}
let Field::Present(parent_rpc) = self.context.rpc_id else {
return Err(F::Rpc);
};
let child_rpc = call.rpc_correlation_id().ok_or(F::Rpc)?.as_str();
let parent_rpc =
std::str::from_utf8(&parent_rpc.bytes[..parent_rpc.len]).map_err(|_| F::Rpc)?;
let sequence = child_rpc
.strip_prefix(parent_rpc)
.and_then(|suffix| suffix.strip_prefix('.'))
.ok_or(F::Rpc)?;
if sequence.is_empty() || !sequence.bytes().all(|b| b.is_ascii_digit()) {
return Err(F::Rpc);
}
let Field::Present(parent_span) = self.context.span_id else {
return Err(F::Span);
};
if parent_span == Span(call.span_id().as_u64()) {
return Err(F::Span);
}
let child = Projection::existing(call, event);
let mut context = self.context;
context.application = child.application;
context.module = child.module;
context.service = child.service;
context.operation = child.operation;
context.rpc_id = child.rpc_id;
context.span_id = child.span_id;
context.route = child.route;
context.attempt = child.attempt;
Ok(RequestDiagnosticScope {
output: self.output,
context,
})
}
pub fn with_phase(mut self, phase: DiagnosticRequestPhase) -> Self {
self.context.lifecycle = Field::Present(SafeText::metadata(phase.as_str()));
self
}
pub fn with_missing(
mut self,
field: DiagnosticContextField,
reason: DiagnosticContextMissing,
) -> Self {
fn set<T>(field: &mut Field<T>, reason: DiagnosticContextMissing) {
if !matches!(field, Field::Present(_)) {
*field = match reason {
DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
DiagnosticContextMissing::Unavailable => Field::Unavailable,
};
}
}
match field {
DiagnosticContextField::Application => set(&mut self.context.application, reason),
DiagnosticContextField::Module => set(&mut self.context.module, reason),
DiagnosticContextField::Service => set(&mut self.context.service, reason),
DiagnosticContextField::Operation => set(&mut self.context.operation, reason),
DiagnosticContextField::DbOperation => set(&mut self.context.db_operation, reason),
DiagnosticContextField::Trace => set(&mut self.context.trace_id, reason),
DiagnosticContextField::Rpc => set(&mut self.context.rpc_id, reason),
DiagnosticContextField::Span => set(&mut self.context.span_id, reason),
DiagnosticContextField::Request => set(&mut self.context.request, reason),
DiagnosticContextField::Route => set(&mut self.context.route, reason),
DiagnosticContextField::Attempt => set(&mut self.context.attempt, reason),
DiagnosticContextField::Scope => set(&mut self.context.scope, reason),
DiagnosticContextField::Task => set(&mut self.context.task, reason),
DiagnosticContextField::Lifecycle => set(&mut self.context.lifecycle, reason),
DiagnosticContextField::Zone => set(&mut self.context.zone, reason),
DiagnosticContextField::Target => set(&mut self.context.target, reason),
}
self
}
pub fn live_db_scope(
output: Option<&'a EmergencyDiagnosticHandle>,
projection: &saddle_core::DbScopeDiagnosticContext<(&CallContext, &EventContext)>,
) -> Self {
let ((call, event), scope) = projection.diagnostic_context();
let mut bound = match output {
Some(output) => Self::established(output, call, event),
None => Self::output_unavailable(call, event),
};
bound.context.scope = Field::Present(scope);
bound
}
pub fn with_db_operation(mut self, operation: DiagnosticDbOperation) -> Self {
self.set_db_operation(operation);
self
}
pub fn set_db_operation(&mut self, operation: DiagnosticDbOperation) {
self.context.db_operation = Field::Present(operation.0);
}
pub fn with_db_operation_missing(mut self, reason: DiagnosticContextMissing) -> Self {
if !matches!(self.context.db_operation, Field::Present(_)) {
self.context.db_operation = match reason {
DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
DiagnosticContextMissing::Unavailable => Field::Unavailable,
};
}
self
}
pub fn with_zone(mut self, zone: DiagnosticZone) -> Self {
self.context.zone = Field::Present(zone.0);
self
}
pub fn with_zone_missing(mut self, reason: DiagnosticContextMissing) -> Self {
if !matches!(self.context.zone, Field::Present(_)) {
self.context.zone = match reason {
DiagnosticContextMissing::NotApplicable => Field::NotApplicable,
DiagnosticContextMissing::NotEstablished => Field::NotEstablished,
DiagnosticContextMissing::Unavailable => Field::Unavailable,
};
}
self
}
pub fn db_scope(
output: &'a EmergencyDiagnosticHandle,
observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
) -> Self {
let ((_, call, event), scope) = observation.diagnostic_context();
let mut bound = Self::established(output, call, event);
bound.context.scope = Field::Present(scope);
bound
}
pub fn db_scope_output_unavailable(
observation: &saddle_core::DbScopeObservation<(crate::Observer, CallContext, EventContext)>,
) -> Self {
let ((_, call, event), scope) = observation.diagnostic_context();
let mut bound = Self::output_unavailable(call, event);
bound.context.scope = Field::Present(scope);
bound
}
pub fn established(
output: &'a EmergencyDiagnosticHandle,
call: &CallContext,
event: &EventContext,
) -> Self {
Self {
output: Some(output),
context: Projection::existing(call, event),
}
}
pub fn output_unavailable(call: &CallContext, event: &EventContext) -> Self {
Self {
output: None,
context: Projection::existing(call, event),
}
}
pub fn capture_required(&self, diagnostic: BoundedDiagnostic) -> RequestSourceReceipt {
self.capture_error((), diagnostic)
}
pub fn capture_existing(
&self,
diagnostic: saddle_core::Diagnostic,
observer: Option<&crate::Observer>,
) -> ExistingDiagnosticReceipt {
let occurrence = diagnostic.occurrence();
let record = Record {
event: "framework.diagnostic",
timestamp_unix_ms: timestamp(),
context: &self.context,
diagnostic: Some(&diagnostic),
diagnostic_reference: occurrence,
axes: None,
source_submission: None,
};
let submission = self
.output
.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
output.submit_existing_record(&record, diagnostic.deferred_stack())
});
if let Some(observer) = observer {
observer.mirror_existing_record(&record, diagnostic.category());
}
FrameworkRequestFailure {
error: (),
context: self.context,
occurrence,
submission,
diagnostic,
}
}
#[track_caller]
pub fn fail<E>(
&self,
error: E,
category: DiagnosticCategory,
cause: BoundedDiagnosticCause,
) -> FrameworkRequestFailure<E> {
let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
self.capture_error(error, diagnostic)
}
fn capture_error<E>(
&self,
error: E,
diagnostic: BoundedDiagnostic,
) -> FrameworkRequestFailure<E> {
let occurrence = diagnostic.occurrence();
let submission = self
.output
.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
output.submit_fixed_record(&Record {
event: "framework.diagnostic",
timestamp_unix_ms: timestamp(),
context: &self.context,
diagnostic: Some(&diagnostic),
diagnostic_reference: occurrence,
axes: None,
source_submission: None,
})
});
FrameworkRequestFailure {
error,
context: self.context,
occurrence,
submission,
diagnostic,
}
}
}
#[must_use = "retain the source receipt with the technical result until its declared boundary"]
pub struct FrameworkRequestFailure<E, D = BoundedDiagnostic> {
error: E,
context: Projection,
occurrence: DiagnosticOccurrence,
submission: DiagnosticSubmission,
diagnostic: D,
}
pub type RequestSourceReceipt = FrameworkRequestFailure<()>;
pub type ExistingDiagnosticReceipt = FrameworkRequestFailure<(), saddle_core::Diagnostic>;
impl<E, D> FrameworkRequestFailure<E, D> {
pub fn error(&self) -> &E {
&self.error
}
pub fn submission(&self) -> DiagnosticSubmission {
self.submission
}
pub fn map_error<F>(self, map: impl FnOnce(E) -> F) -> FrameworkRequestFailure<F, D> {
FrameworkRequestFailure {
error: map(self.error),
context: self.context,
occurrence: self.occurrence,
submission: self.submission,
diagnostic: self.diagnostic,
}
}
pub fn record_boundary(
&self,
output: &EmergencyDiagnosticHandle,
axes: &DiagnosticOutcomeAxes,
) -> DiagnosticSubmission {
self.record_boundary_optional(Some(output), axes)
}
pub fn record_boundary_optional(
&self,
output: Option<&EmergencyDiagnosticHandle>,
axes: &DiagnosticOutcomeAxes,
) -> DiagnosticSubmission {
output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
output.submit_fixed_record(&Record {
event: "framework.boundary.outcome",
timestamp_unix_ms: timestamp(),
context: &self.context,
diagnostic: None::<&BoundedDiagnostic>,
diagnostic_reference: self.occurrence,
axes: Some(axes),
source_submission: Some(self.submission),
})
})
}
pub fn finish_boundary_retained(
self,
output: Option<&EmergencyDiagnosticHandle>,
axes: &DiagnosticOutcomeAxes,
) -> (Self, BoundaryDiagnosticDelivery) {
let delivery = BoundaryDiagnosticDelivery {
source: self.submission,
boundary: self.record_boundary_optional(output, axes),
occurrence: self.occurrence,
};
(self, delivery)
}
pub fn source_diagnostic(&self) -> &D {
&self.diagnostic
}
pub fn finish_boundary(
self,
output: &EmergencyDiagnosticHandle,
axes: &DiagnosticOutcomeAxes,
) -> (E, BoundaryDiagnosticDelivery) {
let boundary = self.record_boundary(output, axes);
(
self.error,
BoundaryDiagnosticDelivery {
source: self.submission,
boundary,
occurrence: self.occurrence,
},
)
}
}
impl<D> FrameworkRequestFailure<(), D> {
pub fn into_reference(self) -> RequestBoundaryReference<D> {
RequestBoundaryReference {
context: self.context,
occurrence: self.occurrence,
submission: self.submission,
diagnostic: self.diagnostic,
}
}
}
#[must_use = "retain source facts and submission status through terminal consumption"]
pub struct RequestBoundaryReference<D = BoundedDiagnostic> {
context: Projection,
occurrence: DiagnosticOccurrence,
submission: DiagnosticSubmission,
diagnostic: D,
}
impl<D> RequestBoundaryReference<D> {
pub(crate) fn record_stage(
&self,
output: Option<&EmergencyDiagnosticHandle>,
axes: &DiagnosticOutcomeAxes,
stage: &'static str,
elapsed_ms: u64,
) -> BoundaryDiagnosticDelivery {
#[derive(Serialize)]
struct StageRecord<'a> {
#[serde(flatten)]
record: Record<'a>,
stage: &'static str,
elapsed_ms: u64,
}
let boundary = output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
output.submit_fixed_record(&StageRecord {
record: Record {
event: "framework.boundary.outcome",
timestamp_unix_ms: timestamp(),
context: &self.context,
diagnostic: None,
diagnostic_reference: self.occurrence,
axes: Some(axes),
source_submission: Some(self.submission),
},
stage,
elapsed_ms,
})
});
BoundaryDiagnosticDelivery {
source: self.submission,
boundary,
occurrence: self.occurrence,
}
}
pub fn occurrence(&self) -> DiagnosticOccurrence {
self.occurrence
}
pub fn source_submission(&self) -> DiagnosticSubmission {
self.submission
}
pub fn source_diagnostic(&self) -> &D {
&self.diagnostic
}
pub fn record(
&self,
output: &EmergencyDiagnosticHandle,
axes: &DiagnosticOutcomeAxes,
) -> DiagnosticSubmission {
self.record_optional(Some(output), axes)
}
pub fn record_optional(
&self,
output: Option<&EmergencyDiagnosticHandle>,
axes: &DiagnosticOutcomeAxes,
) -> DiagnosticSubmission {
output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
output.submit_fixed_record(&Record {
event: "framework.boundary.outcome",
timestamp_unix_ms: timestamp(),
context: &self.context,
diagnostic: None::<&BoundedDiagnostic>,
diagnostic_reference: self.occurrence,
axes: Some(axes),
source_submission: Some(self.submission),
})
})
}
pub fn finish_retained(
self,
output: Option<&EmergencyDiagnosticHandle>,
axes: &DiagnosticOutcomeAxes,
) -> (Self, BoundaryDiagnosticDelivery) {
let delivery = BoundaryDiagnosticDelivery {
source: self.submission,
boundary: self.record_optional(output, axes),
occurrence: self.occurrence,
};
(self, delivery)
}
}
pub struct BoundaryDiagnosticDelivery {
source: DiagnosticSubmission,
boundary: DiagnosticSubmission,
occurrence: DiagnosticOccurrence,
}
impl BoundaryDiagnosticDelivery {
pub fn source_submission(&self) -> DiagnosticSubmission {
self.source
}
pub fn boundary_submission(&self) -> DiagnosticSubmission {
self.boundary
}
pub fn occurrence(&self) -> DiagnosticOccurrence {
self.occurrence
}
}
#[derive(Serialize)]
struct Record<'a, D = BoundedDiagnostic> {
event: &'static str,
timestamp_unix_ms: u128,
context: &'a Projection,
diagnostic: Option<&'a D>,
diagnostic_reference: DiagnosticOccurrence,
axes: Option<&'a DiagnosticOutcomeAxes>,
#[serde(skip_serializing_if = "Option::is_none")]
source_submission: Option<DiagnosticSubmission>,
}
fn timestamp() -> u128 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
}
#[cfg(test)]
mod early_tests {
use super::*;
use saddle_core::*;
fn ok<T>(value: std::result::Result<T, ContextBindingError<T>>) -> T {
match value {
Ok(value) => value,
Err(_) => panic!("unexpected context conflict"),
}
}
fn cause() -> BoundedDiagnosticCause {
BoundedDiagnosticCause::new(
DiagnosticStage::RequestDecode,
DiagnosticCode::new("request.read_failed").unwrap(),
)
}
fn context(scope: &RequestDiagnosticScope<'_>) -> serde_json::Value {
serde_json::to_value(scope.context).unwrap()
}
#[test]
fn early_absence_and_output_are_independent_and_receipt_retains_error() {
let scope =
RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("app"))
.with_missing(
DiagnosticContextField::DbOperation,
DiagnosticContextMissing::NotApplicable,
)
.with_missing(
DiagnosticContextField::Application,
DiagnosticContextMissing::Unavailable,
)
.with_phase(DiagnosticRequestPhase::ReadingHead);
let value = context(&scope);
assert_eq!(value["application"]["value"]["value"], "app");
for field in ["trace_id", "rpc_id", "span_id", "request", "scope", "task"] {
assert_eq!(value[field]["state"], "not_established");
assert!(value[field].get("value").is_none());
}
assert_eq!(value["db_operation"]["state"], "not_applicable");
assert_eq!(value["lifecycle"]["value"]["value"], "reading_head");
let failure = scope.fail(17u32, DiagnosticCategory::UnexpectedError, cause());
let id = failure.source_diagnostic().id();
assert_eq!(
failure.submission(),
DiagnosticSubmission::OutputUnavailable
);
let mapped = failure.map_error(|code| code + 1);
let (retained, delivery) =
mapped.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
assert_eq!(*retained.error(), 18);
assert_eq!(retained.source_diagnostic().id(), id);
assert_eq!(
delivery.source_submission(),
DiagnosticSubmission::OutputUnavailable
);
assert_eq!(
delivery.boundary_submission(),
DiagnosticSubmission::OutputUnavailable
);
let lost = RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable());
assert_eq!(context(&lost)["trace_id"]["state"], "unavailable");
}
#[test]
fn known_partial_identity_atomic_conflict_and_promotion_preserve_snapshots() {
let trace = TraceCorrelationId::new("gateway-opaque-原值").unwrap();
let foreign = TraceCorrelationId::new("foreign").unwrap();
let early = ok(EarlyRequestContext::socket_accepted("app").with_trace(&trace));
let early = match early.with_trace(&foreign) {
Ok(_) => panic!("foreign trace accepted"),
Err(error) => {
assert_eq!(error.field(), DiagnosticContextField::Trace);
error.into_original()
}
};
let request = crate::RequestIdentity::new("request-1").unwrap();
let early = ok(early.with_request(&request));
let scope = ok(RequestDiagnosticScope::early(None, early)
.with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
let before = context(&scope);
let receipt = scope
.fail((), DiagnosticCategory::UnexpectedError, cause())
.into_reference();
let call = CallContext::new(
"app".into(),
"module".into(),
"service".into(),
"operation".into(),
TraceId::from_u128(1),
SpanId::from_u64(2),
)
.with_trace_correlation_id(foreign);
let event =
EventContext::new(request, crate::RouteIdentity::new("/route").unwrap(), 1).unwrap();
let scope = match scope.bind_established(&call, &event) {
Ok(_) => panic!("foreign promotion accepted"),
Err(error) => error.into_original(),
};
assert_eq!(context(&scope), before); let call = call.with_trace_correlation_id(trace);
let scope = ok(scope.bind_established(&call, &event));
let after = context(&scope);
assert_eq!(after["trace_id"]["value"], "gateway-opaque-原值");
assert_eq!(after["task"]["value"]["value"], "42");
assert_eq!(after["span_id"]["value"], "0000000000000002");
assert_eq!(serde_json::to_value(receipt.context).unwrap(), before);
assert!(
scope.with_output(None).context.request
== Field::Present(SafeText::metadata("request-1"))
);
}
#[test]
fn task_projection_is_bounded_and_cannot_be_replaced() {
for unsafe_id in ["", "request_task", "-1", "https://secret", "12\n3"] {
assert!(DiagnosticTaskId::from_runtime_id(unsafe_id).is_none());
}
assert!(DiagnosticTaskId::from_runtime_id(&"1".repeat(65)).is_none());
let scope = ok(RequestDiagnosticScope::early(
None,
EarlyRequestContext::socket_accepted("app"),
)
.with_task(DiagnosticTaskId::from_runtime_id("42").unwrap()));
let original = context(&scope);
let scope = match scope.with_task(DiagnosticTaskId::from_runtime_id("43").unwrap()) {
Ok(_) => panic!("changed task identity"),
Err(error) => error.into_original(),
};
assert_eq!(context(&scope), original);
}
}
#[cfg(test)]
mod outbound_child_tests {
use super::*;
use saddle_core::*;
fn event(request: &str, route: &str) -> EventContext {
EventContext::new(
crate::RequestIdentity::new(request).unwrap(),
crate::RouteIdentity::new(route).unwrap(),
1,
)
.unwrap()
}
fn call(trace: &str, rpc: &str, span: u64) -> CallContext {
CallContext::new(
"saddle".into(),
"zone-a".into(),
"profusecontract".into(),
"invoke".into(),
TraceId::from_u128(1),
SpanId::from_u64(span),
)
.with_trace_correlation_id(TraceCorrelationId::new(trace).unwrap())
.with_rpc_correlation_id(RpcCorrelationId::new(rpc))
}
fn cause() -> BoundedDiagnosticCause {
BoundedDiagnosticCause::new(
DiagnosticStage::RequestDecode,
DiagnosticCode::new("outbound.connect_failed").unwrap(),
)
}
#[test]
fn outbound_child_preserves_live_scope_and_receipts() {
let parent = call("opaque-trace", "0.4", 1);
let child = call("opaque-trace", "0.4.1", 2);
let parent_event = event("request-1", "/incoming");
let child_event = event("request-1", "remote.function");
let (_, issuer) = DbPhysicalDispositionIssuer::issue().into_startup_and_request_issuer();
let (request, execution) = issuer.issue_request().unwrap();
let live = request
.project_diagnostic_context(&execution, (&parent, &parent_event))
.ok()
.unwrap();
let scope = RequestDiagnosticScope::live_db_scope(None, &live)
.with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
.unwrap_or_else(|_| panic!("task"))
.with_zone(DiagnosticZone::from_validated_ingress("zone-a").unwrap())
.with_db_operation(DiagnosticDbOperation::from_registered("orders.query").unwrap());
let original = serde_json::to_value(scope.context).unwrap();
let old = scope.fail(7, DiagnosticCategory::UnexpectedError, cause());
let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
let value = serde_json::to_value(derived.context).unwrap();
for field in [
"request",
"trace_id",
"task",
"zone",
"scope",
"db_operation",
"lifecycle",
"target",
] {
assert_eq!(value[field], original[field], "{field}");
}
assert_eq!(value["rpc_id"]["value"], "0.4.1");
assert_eq!(value["span_id"]["value"], "0000000000000002");
assert_eq!(value["route"]["value"]["value"], "remote.function");
assert_eq!(serde_json::to_value(scope.context).unwrap(), original);
assert_eq!(serde_json::to_value(old.context).unwrap(), original);
assert!(
scope
.reborrow()
.bind_established(&child, &child_event)
.is_err()
);
let source = derived.fail(9, DiagnosticCategory::UnexpectedError, cause());
let id = source.source_diagnostic().id();
let (retained, delivery) =
source.finish_boundary_retained(None, &DiagnosticOutcomeAxes::default());
assert_eq!(retained.source_diagnostic().id(), id);
assert_eq!(*retained.error(), 9);
assert_eq!(
delivery.source_submission(),
DiagnosticSubmission::OutputUnavailable
);
assert_eq!(
delivery.boundary_submission(),
DiagnosticSubmission::OutputUnavailable
);
assert_eq!(serde_json::to_value(retained.context).unwrap(), value);
}
#[test]
fn outbound_child_rejects_foreign_missing_and_redaction_collisions() {
let parent = call("trace", "0", 1);
let scope = RequestDiagnosticScope::output_unavailable(&parent, &event("a@b", "/in"));
let before = serde_json::to_value(scope.context).unwrap();
let child_event = event("a@b", "remote");
for (child, ev, expected) in [
(
call("foreign", "0.1", 2),
child_event.clone(),
DiagnosticContextField::Trace,
),
(
call("trace", "0.1", 2),
event("c@d", "remote"),
DiagnosticContextField::Request,
),
(
call("trace", "01.1", 2),
child_event.clone(),
DiagnosticContextField::Rpc,
),
(
call("trace", "0", 2),
child_event.clone(),
DiagnosticContextField::Rpc,
),
(
call("trace", "0.1.2", 2),
child_event.clone(),
DiagnosticContextField::Rpc,
),
(
call("trace", "0.x", 2),
child_event.clone(),
DiagnosticContextField::Rpc,
),
(
call("trace", "0.1", 1),
child_event.clone(),
DiagnosticContextField::Span,
),
] {
assert_eq!(
scope.derive_outbound_child(&child, &ev).err(),
Some(expected)
);
assert_eq!(serde_json::to_value(scope.context).unwrap(), before);
}
let child = call("trace", "0.1", 2);
let derived = scope.derive_outbound_child(&child, &child_event).unwrap();
let json = serde_json::to_string(&derived.context).unwrap();
assert!(!json.contains("a@b") && !json.contains("request_binding"));
assert!(
RequestDiagnosticScope::early(None, EarlyRequestContext::unavailable())
.derive_outbound_child(&child, &child_event)
.is_err()
);
let missing_rpc = parent.clone().with_rpc_correlation_id(None);
assert_eq!(
RequestDiagnosticScope::output_unavailable(&missing_rpc, &child_event)
.derive_outbound_child(&child, &child_event)
.err(),
Some(DiagnosticContextField::Rpc)
);
assert!(
scope
.reborrow()
.bind_request_identity(&crate::RequestIdentity::new("c@d").unwrap())
.is_err()
);
assert!(
EarlyRequestContext::unavailable()
.with_request(&crate::RequestIdentity::new("a@b").unwrap())
.unwrap_or_else(|_| panic!("first identity"))
.with_request(&crate::RequestIdentity::new("c@d").unwrap())
.is_err()
);
}
}