saddle_observability/
root_diagnostic.rs

1//! Single-root consumer. Legacy snapshot APIs remain for not-yet-migrated domains.
2use crate::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext, Observer};
3use saddle_core::{
4    BoundedDiagnostic, ContextFact, Diagnostic, DiagnosticCategory, DiagnosticCode,
5    DiagnosticOccurrence, DiagnosticOutcomeAxes, RequestExecutionView,
6};
7use serde::Serialize;
8pub(crate) mod original;
9pub(crate) use original::{stream_record, Segment as EncodedSegment, OriginalOutput as SegmentOutput};
10mod terminal;
11pub use terminal::{RecordedTerminalFailure, WrittenTerminalOriginal};
12pub use original::{request_terminal_audit_error, request_terminal_audit_disposition,
13    request_terminal_audit_passed, request_terminal_audit_cleanup};
14pub use original::{
15    GrpcTargetContext, GrpcStatusFacts, OriginalCaptureState, RemoteProtocolContext, RemoteTargetContext, UnrootedCaptureFacts, UnrootedDiagnosticScope,
16    WrittenComponentFailure, WrittenUnrootedFailure,
17    ComponentSourceKind, ComponentSourceReceipt, original_capture_layout,
18    database_background_error, database_transaction_error, database_operation_error,
19    database_operation_borrowed_error, database_operation_borrowed_input_bounded,
20    write_json_error_debug,
21    process_startup_error, process_config_error,
22    process_finalization_error, process_component_start_error, process_component_cleanup_error,
23    process_component_start_recorded, process_component_cleanup_recorded,
24    process_shutdown_signal_error, process_request_drain_error, process_post_driver_error,
25    process_bootstrap_error,
26    service_registry_source_error, service_registry_source_description, service_route_source_description,
27    service_entry_trace_source_error, ServiceRegistrySourceOutput,
28    service_production_fact_source_error, service_compiled_build_source_error,
29    service_request_source_error, service_request_source_description,
30    runtime_admission_source_description,
31    runtime_unrooted_admission_source_description,
32};
33
34/// A safe public projection paired with the writer's sealed confirmation of
35/// the same original occurrence. Formal consumers can require this type
36/// instead of accepting a freely constructed Core error as source evidence.
37#[must_use = "pass the recorded failure to its formal consumer"]
38pub struct RecordedSaddleError {
39    safe: saddle_core::SaddleError,
40    written: WrittenSource,
41}
42
43enum WrittenSource {
44    Unrooted(WrittenUnrootedFailure),
45    Component(WrittenComponentFailure),
46}
47impl WrittenSource {
48    fn occurrence(&self) -> DiagnosticOccurrence {
49        match self {
50            Self::Unrooted(written) => written.occurrence(),
51            Self::Component(written) => written.occurrence(),
52        }
53    }
54}
55
56impl RecordedSaddleError {
57    pub(crate) fn previously_written_writer_error<E>(
58        source: E,
59        diagnostic: Diagnostic,
60        written: Option<WrittenComponentFailure>,
61    ) -> LifecycleSourceFailure
62    where E: std::error::Error + Send + Sync + 'static {
63        let safe = saddle_core::SaddleError::new(
64            saddle_core::ErrorKind::Infrastructure,
65            "observability.shutdown_failed",
66            "structured log shutdown failed",
67        ).with_diagnostic(diagnostic);
68        match written {
69            Some(written) => Self::from_component_written(safe, written)
70                .map_or_else(|safe| UnconfirmedSaddleSource { safe, source }.erase(),
71                    LifecycleSourceFailure::from),
72            None => UnconfirmedSaddleSource {
73                safe: safe.with_unconfirmed_source(), source,
74            }.erase(),
75        }
76    }
77
78    /// Record the owned source before running the public safety projection.
79    /// The selected output is mandatory; a failed write returns the original
80    /// to its owner and never creates a positive credential.
81    pub fn capture_unrooted<E, P>(
82        source: E,
83        diagnostic: Diagnostic,
84        application: &saddle_core::ContextLabel,
85        lifecycle: saddle_core::RequestViewPhase,
86        output: &crate::SourceOutput,
87        stage: RootRequestEvent,
88        outcome: RootOutcomeFacts,
89        project: P,
90    ) -> Result<Self, UnconfirmedSaddleSource<E>>
91    where
92        E: std::error::Error + Send + Sync + 'static,
93        P: FnOnce(&E, Diagnostic) -> saddle_core::SaddleError,
94    {
95        let scope = UnrootedDiagnosticScope::new(application, lifecycle, Some(output.handle()));
96        let written = scope.source_existing_error_recorded(&source, &diagnostic, stage, outcome);
97        let safe = project(&source, diagnostic);
98        match written {
99            Ok(written) => Self::from_written(safe, written)
100                .map_err(|safe| UnconfirmedSaddleSource { safe, source }),
101            Err(_) => Err(UnconfirmedSaddleSource {
102                safe: safe.with_unconfirmed_source(), source,
103            }),
104        }
105    }
106
107    /// Component startup and cleanup use distinct sealed writer credentials.
108    pub(crate) fn capture_component<E, P>(
109        source: E,
110        diagnostic: Diagnostic,
111        application: ContextFact<saddle_core::ContextLabel>,
112        output: &crate::SourceOutput,
113        kind: ComponentSourceKind,
114        project: P,
115    ) -> Result<Self, UnconfirmedSaddleSource<E>>
116    where
117        E: std::error::Error + Send + Sync + 'static,
118        P: FnOnce(&E, Diagnostic) -> saddle_core::SaddleError,
119    {
120        let written = match kind {
121            ComponentSourceKind::Start => process_component_start_recorded(
122                Some(output.handle()), application, &diagnostic, &source),
123            ComponentSourceKind::Cleanup => process_component_cleanup_recorded(
124                Some(output.handle()), application, &diagnostic, &source),
125        };
126        let safe = project(&source, diagnostic);
127        match written {
128            Ok(written) => Self::from_component_written(safe, written)
129                .map_err(|safe| UnconfirmedSaddleSource { safe, source }),
130            Err(_) => Err(UnconfirmedSaddleSource {
131                safe: safe.with_unconfirmed_source(), source,
132            }),
133        }
134    }
135
136    /// Consume a provider Result directly. Its Err is still owned when the
137    /// writer sees it; callers supply only static safe classification, so
138    /// they cannot run a projection callback before the source write.
139    pub fn component_result<T, E>(
140        result: Result<T, E>,
141        application: &saddle_core::ContextLabel,
142        output: &crate::SourceOutput,
143        kind: ComponentSourceKind,
144        primary: Option<DiagnosticOccurrence>,
145        safe_kind: saddle_core::ErrorKind,
146        code: &'static str,
147        safe_message: &'static str,
148    ) -> Result<T, LifecycleSourceFailure>
149    where E: std::error::Error + Send + Sync + 'static {
150        result.map_err(|raw| {
151            let source_stage = match kind {
152                ComponentSourceKind::Start => saddle_core::DiagnosticStage::StartupListener,
153                ComponentSourceKind::Cleanup => saddle_core::DiagnosticStage::ShutdownComponent,
154            };
155            let mut diagnostic = Diagnostic::capture(
156                DiagnosticCategory::UnexpectedError,
157                saddle_core::CaptureSite::FirstObserved,
158                saddle_core::DiagnosticCause::new(source_stage,
159                    DiagnosticCode::new(code).expect("static component code")),
160            );
161            if let Some(primary) = primary.as_ref() {
162                diagnostic = diagnostic.during_cleanup_of_occurrence(primary);
163            }
164            Self::capture_component(raw, diagnostic,
165                ContextFact::Present(application.clone()), output, kind,
166                |_, diagnostic| saddle_core::SaddleError::new(
167                    safe_kind, code, safe_message).with_diagnostic(diagnostic),
168            ).map_or_else(UnconfirmedSaddleSource::erase, LifecycleSourceFailure::from)
169        })
170    }
171
172    pub(crate) fn from_written(
173        safe: saddle_core::SaddleError,
174        written: WrittenUnrootedFailure,
175    ) -> Result<Self, saddle_core::SaddleError> {
176        Self::pair(safe, WrittenSource::Unrooted(written))
177    }
178
179    pub(crate) fn from_component_written(
180        safe: saddle_core::SaddleError,
181        written: WrittenComponentFailure,
182    ) -> Result<Self, saddle_core::SaddleError> {
183        Self::pair(safe, WrittenSource::Component(written))
184    }
185
186    fn pair(safe: saddle_core::SaddleError, written: WrittenSource)
187        -> Result<Self, saddle_core::SaddleError> {
188        if safe.diagnostic().is_some_and(|diagnostic|
189            written.occurrence().matches_diagnostic(diagnostic)) {
190            Ok(Self { safe, written })
191        } else {
192            Err(safe.with_unconfirmed_source())
193        }
194    }
195
196    pub fn safe(&self) -> &saddle_core::SaddleError { &self.safe }
197
198    pub fn component_kind(&self) -> Option<ComponentSourceKind> {
199        match &self.written {
200            WrittenSource::Component(written) => Some(written.kind()),
201            WrittenSource::Unrooted(_) => None,
202        }
203    }
204}
205
206/// A failed source write retains the actual owned error alongside the safe
207/// public projection. It has no conversion into a written failure.
208#[must_use = "propagate the unconfirmed source to the formal failure consumer"]
209pub struct UnconfirmedSaddleSource<E> {
210    safe: saddle_core::SaddleError,
211    source: E,
212}
213
214impl<E> UnconfirmedSaddleSource<E> {
215    pub fn safe(&self) -> &saddle_core::SaddleError { &self.safe }
216    pub fn source(&self) -> &E { &self.source }
217
218    pub fn erase(self) -> LifecycleSourceFailure
219    where E: std::error::Error + Send + Sync + 'static {
220        LifecycleSourceFailure(LifecycleSourceFailureState::Unconfirmed {
221            safe: self.safe,
222            source: Box::new(self.source),
223        })
224    }
225}
226
227/// A formal component result carries either a sealed full write or the
228/// original error that could not be confirmed. Neither variant can be
229/// manufactured from a bare Core error or an arbitrary Any receipt.
230#[must_use = "forward the lifecycle failure to Application"]
231pub struct LifecycleSourceFailure(LifecycleSourceFailureState);
232enum LifecycleSourceFailureState {
233    Written(RecordedSaddleError),
234    WrongPhase {
235        safe: saddle_core::SaddleError,
236        _recorded: RecordedSaddleError,
237    },
238    Unconfirmed {
239        safe: saddle_core::SaddleError,
240        source: Box<dyn std::error::Error + Send + Sync>,
241    },
242}
243impl LifecycleSourceFailure {
244    pub fn safe(&self) -> &saddle_core::SaddleError {
245        match &self.0 {
246            LifecycleSourceFailureState::Written(recorded) => recorded.safe(),
247            LifecycleSourceFailureState::WrongPhase { safe, .. } => safe,
248            LifecycleSourceFailureState::Unconfirmed { safe, .. } => safe,
249        }
250    }
251
252    /// Project at the Application result boundary. A failed source write
253    /// carries its owned original through the public result as negative evidence.
254    pub fn into_safe(self) -> saddle_core::SaddleError {
255        match self.0 {
256            LifecycleSourceFailureState::Written(recorded) => recorded.safe,
257            LifecycleSourceFailureState::WrongPhase { safe, _recorded } =>
258                safe.with_unconfirmed_original(Box::new(_recorded)),
259            LifecycleSourceFailureState::Unconfirmed { safe, source } =>
260                safe.with_unconfirmed_original(source),
261        }
262    }
263
264    /// A credential for another lifecycle phase cannot confirm this phase.
265    /// Keep the already written original owned until the caller handles the
266    /// rejected result; never turn it into a positive credential here.
267    pub fn require_component_kind(self, expected: ComponentSourceKind) -> Self {
268        match self.0 {
269            LifecycleSourceFailureState::Written(recorded)
270                if recorded.component_kind() != Some(expected) => {
271                let safe = saddle_core::SaddleError::new(
272                    saddle_core::ErrorKind::Infrastructure,
273                    "component.source_phase_mismatch",
274                    "component source was recorded for another lifecycle phase",
275                ).with_unconfirmed_source();
276                Self(LifecycleSourceFailureState::WrongPhase { safe, _recorded: recorded })
277            }
278            state => Self(state),
279        }
280    }
281
282    pub fn original_if_unconfirmed(&self) -> Option<&(dyn std::error::Error + Send + Sync)> {
283        match &self.0 {
284            LifecycleSourceFailureState::Written(_) => None,
285            LifecycleSourceFailureState::WrongPhase { .. } => None,
286            LifecycleSourceFailureState::Unconfirmed { source, .. } => Some(source.as_ref()),
287        }
288    }
289
290}
291impl From<RecordedSaddleError> for LifecycleSourceFailure {
292    fn from(recorded: RecordedSaddleError) -> Self {
293        Self(LifecycleSourceFailureState::Written(recorded))
294    }
295}
296
297#[cfg(test)]
298mod lifecycle_projection_tests {
299    use super::*;
300
301    #[test]
302    fn unconfirmed_application_projection_retains_owned_original_without_receipt() {
303        let safe = saddle_core::SaddleError::new(saddle_core::ErrorKind::Infrastructure,
304            "component.unconfirmed", "safe").with_unconfirmed_source();
305        let failure = LifecycleSourceFailure(LifecycleSourceFailureState::Unconfirmed {
306            safe, source: Box::new(std::io::Error::other("M_PRIVATE_ORIGINAL")),
307        });
308        let projected = failure.into_safe();
309        assert!(projected.source_unavailable());
310        assert!(projected.source_receipt::<ComponentSourceReceipt>().is_none());
311        assert_eq!(projected.unconfirmed_original().unwrap().to_string(), "M_PRIVATE_ORIGINAL");
312        assert!(!projected.to_string().contains("M_PRIVATE_ORIGINAL"));
313    }
314}
315
316pub type RecordedLifecycleFuture<'a> = std::pin::Pin<Box<dyn std::future::Future<
317    Output = Result<(), LifecycleSourceFailure>> + Send + 'a>>;
318
319/// Formal components borrow the selected process output for every fallible
320/// phase. Legacy Core lifecycle results do not implement this contract.
321pub trait RecordedComponentLifecycle: Send + Sync {
322    fn name(&self) -> &'static str;
323    fn start<'a>(&'a self, application: &'a saddle_core::ContextLabel,
324        output: &'a crate::SourceOutput) -> RecordedLifecycleFuture<'a>;
325    fn shutdown<'a>(&'a self, application: &'a saddle_core::ContextLabel,
326        output: &'a crate::SourceOutput)
327        -> RecordedLifecycleFuture<'a>;
328    fn shutdown_with_primary<'a>(
329        &'a self,
330        application: &'a saddle_core::ContextLabel,
331        output: &'a crate::SourceOutput,
332        primary: Option<DiagnosticOccurrence>,
333    ) -> RecordedLifecycleFuture<'a>;
334}
335
336impl std::fmt::Display for RecordedSaddleError {
337    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
338        self.safe.fmt(formatter)
339    }
340}
341impl std::fmt::Debug for RecordedSaddleError {
342    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
343        std::fmt::Debug::fmt(&self.safe, formatter)
344    }
345}
346impl std::error::Error for RecordedSaddleError {
347    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
348        Some(&self.safe)
349    }
350}
351
352/// Read the existing typed call/event; this does not create a second context.
353pub fn request_identity_group(
354    call: &saddle_core::CallContext,
355    event: &EventContext,
356    zone: ContextFact<saddle_core::ContextLabel>,
357) -> Result<saddle_core::RequestIdentityGroup, saddle_core::ContextConflict> {
358    saddle_core::RequestIdentityGroup::from_validated(
359        call,
360        event.diagnostic_request(),
361        event.diagnostic_route(),
362        event.diagnostic_attempt(),
363        zone,
364    )
365}
366/// Consume the exact child Call/Event already established by Boundary.
367pub fn request_child_view(
368    parent: &RequestExecutionView,
369    call: &saddle_core::CallContext,
370    event: &EventContext,
371) -> Result<RequestExecutionView, saddle_core::ContextConflict> {
372    parent.child(
373        call,
374        event.diagnostic_request(),
375        event.diagnostic_route(),
376        event.diagnostic_attempt(),
377    )
378}
379
380/// Closed event families, never a caller-supplied JSON key or arbitrary message.
381#[derive(Clone, Copy, Debug, Serialize)]
382#[serde(rename_all = "snake_case")]
383pub enum RootRequestEvent {
384    Ingress,
385    Admission,
386    Handler,
387    Database,
388    Outbound,
389    Response,
390    Finalization,
391    Supervision,
392}
393
394/// Reported only by the DB owner. Rejected is NOT proof of confirmed rollback.
395#[derive(Clone, Copy, Serialize)]
396#[serde(rename_all = "snake_case")]
397pub enum RequestTransactionFact {
398    Committed,
399    Rejected,
400    Unknown,
401}
402#[derive(Clone, Copy, Serialize)]
403pub struct RootOutcomeFacts {
404    pub axes: DiagnosticOutcomeAxes,
405    pub transaction: ContextFact<RequestTransactionFact>,
406}
407impl Default for RootOutcomeFacts {
408    fn default() -> Self {
409        Self {
410            axes: DiagnosticOutcomeAxes::default(),
411            transaction: ContextFact::Unavailable,
412        }
413    }
414}
415
416#[derive(Serialize)]
417#[serde(untagged)]
418enum SourceDetail<'a> {
419    Bounded(&'a BoundedDiagnostic),
420    Existing(&'a Diagnostic),
421}
422
423fn timestamp() -> u128 {
424    std::time::SystemTime::now()
425        .duration_since(std::time::UNIX_EPOCH)
426        .unwrap_or_default()
427        .as_millis()
428}
429
430#[derive(Serialize)]
431struct RootRecord<'a> {
432    schema_version: u8,
433    timestamp_unix_ms: u128,
434    level: &'static str,
435    elapsed_ms: Option<u64>,
436    event: &'static str,
437    stage: RootRequestEvent,
438    context: &'a RequestExecutionView,
439    source_context: Option<&'a RequestExecutionView>,
440    occurrence: Option<DiagnosticOccurrence>,
441    classification: Option<DiagnosticCode>,
442    source_submission: Option<DiagnosticSubmission>,
443    original_capture: Option<OriginalCaptureState>,
444    axes: RootOutcomeFacts,
445    source_outcome: Option<RootOutcomeFacts>,
446    category: Option<DiagnosticCategory>,
447    detail_status: &'static str,
448    diagnostic: Option<SourceDetail<'a>>,
449}
450
451#[derive(Clone, Copy)]
452struct OrdinaryPayload { stage: RootRequestEvent, elapsed_ms: Option<u64>, axes: RootOutcomeFacts }
453impl OrdinaryPayload {
454    fn transaction_state(&self) -> &'static str {
455        match self.axes.transaction { ContextFact::Present(_) => "present", ContextFact::Unavailable => "unavailable", ContextFact::NotEstablished => "not_established", ContextFact::NotApplicable => "not_applicable" }
456    }
457}
458impl Serialize for OrdinaryPayload {
459    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
460        use serde::ser::SerializeMap;
461        let mut map = s.serialize_map(None)?;
462        map.serialize_entry("stage", &self.stage)?;
463        map.serialize_entry("elapsedMs", &self.elapsed_ms)?;
464        map.serialize_entry("operation", &self.axes.axes.operation)?;
465        map.serialize_entry("physical", &self.axes.axes.physical)?;
466        map.serialize_entry("business", &self.axes.axes.business)?;
467        map.serialize_entry("businessCode", &self.axes.axes.business_code)?;
468        map.serialize_entry("delivery", &self.axes.axes.delivery)?;
469        map.serialize_entry("bytesWritten", &self.axes.axes.bytes_written)?;
470        map.serialize_entry("cleanup", &self.axes.axes.cleanup)?;
471        map.serialize_entry("transactionState", self.transaction_state())?;
472        if let ContextFact::Present(transaction) = self.axes.transaction { map.serialize_entry("transaction", &transaction)?; }
473        map.end()
474    }
475}
476impl crate::ContextRecordPayload for OrdinaryPayload {
477    fn contains(&self, path: &str) -> bool {
478        if path.is_empty() { return true; }
479        let Ok(pointer) = saddle_core::json_pointer::JsonPointer::parse(path) else { return false; };
480        let mut segments = pointer.segments();
481        let Some(key) = segments.next() else { return false; };
482        if segments.next().is_some() { return false; }
483        ["stage", "elapsedMs", "operation", "physical", "business", "businessCode", "delivery", "bytesWritten", "cleanup", "transactionState"].iter().any(|field| key.matches(field)) ||
484            (key.matches("transaction") && matches!(self.axes.transaction, ContextFact::Present(_)))
485    }
486    fn write_selected<W: std::io::Write>(&self, path: &str, writer: &mut W) -> serde_json::Result<()> {
487        if path.is_empty() { return serde_json::to_writer(writer, self); }
488        if !self.contains(path) { return serde_json::to_writer(writer, &()); }
489        let key = saddle_core::json_pointer::JsonPointer::parse(path).unwrap().segments().next().unwrap();
490        macro_rules! field { ($name:literal, $value:expr) => { if key.matches($name) { return serde_json::to_writer(writer, &$value); } }; }
491        field!("stage", self.stage); field!("elapsedMs", self.elapsed_ms);
492        field!("operation", self.axes.axes.operation); field!("physical", self.axes.axes.physical);
493        field!("business", self.axes.axes.business); field!("businessCode", self.axes.axes.business_code);
494        field!("delivery", self.axes.axes.delivery); field!("bytesWritten", self.axes.axes.bytes_written); field!("cleanup", self.axes.axes.cleanup);
495        field!("transactionState", self.transaction_state());
496        if let ContextFact::Present(transaction) = self.axes.transaction { return serde_json::to_writer(writer, &transaction); }
497        serde_json::to_writer(writer, &())
498    }
499}
500
501/// Light, single-consumption source receipt: no exception body, original error,
502/// copied root facts, output handle, task/DB capability or account owner.
503///
504/// ```compile_fail
505/// use saddle_observability::RootRequestFailure;
506/// fn duplicate(f: RootRequestFailure) { let _ = f.clone(); }
507/// ```
508/// ```compile_fail
509/// use saddle_observability::RootRequestFailure;
510/// let f = RootRequestFailure {};
511/// ```
512#[must_use = "carry the source receipt to its declared terminal boundary"]
513pub struct RootRequestFailure {
514    source: RequestExecutionView,
515    occurrence: DiagnosticOccurrence,
516    classification: DiagnosticCode,
517    source_submission: DiagnosticSubmission,
518    category: DiagnosticCategory,
519    outcome: RootOutcomeFacts,
520    original_capture: OriginalCaptureState,
521}
522
523/// A source whose complete terminal record was acknowledged by the file writer.
524/// The private field prevents callers from promoting queued or missing output.
525///
526/// ```compile_fail
527/// use saddle_observability::root_diagnostic::WrittenRootFailure;
528/// fn fake(queued: saddle_observability::RootRequestFailure) -> WrittenRootFailure {
529///     WrittenRootFailure(queued)
530/// }
531/// ```
532#[must_use = "carry the written source to its terminal boundary"]
533pub struct WrittenRootFailure(RootRequestFailure);
534impl WrittenRootFailure {
535    pub fn into_source(self) -> RootRequestFailure { self.0 }
536    pub fn occurrence(&self) -> DiagnosticOccurrence { self.0.occurrence }
537}
538
539/// Root-free public error projection. Holding it cannot keep any request alive.
540#[derive(Clone, Copy, Serialize)]
541pub struct PublicRequestFailure {
542    occurrence: DiagnosticOccurrence,
543    classification: DiagnosticCode,
544    source_submission: DiagnosticSubmission,
545    terminal_submission: DiagnosticSubmission,
546    category: DiagnosticCategory,
547    source_outcome: RootOutcomeFacts,
548    original_capture: OriginalCaptureState,
549}
550
551/// A supervision result cannot be a naked technical code: failure consumes the
552/// original source receipt. Runtime retains its own physical ownership outside it.
553///
554/// ```compile_fail
555/// use saddle_observability::RootSupervisionReturn;
556/// use saddle_core::DiagnosticCode;
557/// let _: RootSupervisionReturn<()> = RootSupervisionReturn::failed(DiagnosticCode::new("task.failed").unwrap());
558/// ```
559/// ```compile_fail
560/// use saddle_observability::{RootSupervisionReturn, RootRequestFailure};
561/// fn replay(f: RootRequestFailure) {
562///     let _first = RootSupervisionReturn::<()>::failed(f);
563/// }
564/// ```
565#[must_use = "the supervisor must consume the actual task result"]
566pub struct RootSupervisionReturn<T> {
567    result: Result<T, RootRequestFailure>,
568}
569impl<T> RootSupervisionReturn<T> {
570    pub fn completed(value: T) -> Self {
571        Self { result: Ok(value) }
572    }
573    pub fn failed(failure: WrittenRootFailure) -> Self {
574        Self {
575            result: Err(failure.into_source()),
576        }
577    }
578    pub fn consume(self) -> Result<T, RootRequestFailure> {
579        self.result
580    }
581}
582
583/// Borrowed logging scope. No output handle or logger ever enters the root.
584pub struct RootDiagnosticScope<'a> {
585    view: &'a RequestExecutionView,
586    output: Option<&'a EmergencyDiagnosticHandle>,
587}
588impl<'a> RootDiagnosticScope<'a> {
589    pub fn new(
590        view: &'a RequestExecutionView,
591        output: Option<&'a EmergencyDiagnosticHandle>,
592    ) -> Self {
593        Self { view, output }
594    }
595    /// Final Runtime-owned ingress deadline; submission is not a write receipt.
596    pub fn ingress_deadline_bound(&self, ingress_deadline_unix_ms: i64) -> DiagnosticSubmission {
597        #[derive(Serialize)]
598        struct DeadlineRecord<'a> {
599            schema_version: u8,
600            event: &'static str,
601            ingress_deadline_unix_ms: i64,
602            context: &'a RequestExecutionView,
603        }
604        self.output.map_or(DiagnosticSubmission::OutputUnavailable, |output|
605            output.submit_fixed_record(&DeadlineRecord {
606                schema_version: 2, event: "ingress_deadline_bound",
607                ingress_deadline_unix_ms, context: self.view,
608            }))
609    }
610
611    /// Detail is consumed and dropped after the one bounded output attempt. Even
612    /// OutputUnavailable/Full/EncodingFailed returns the same lightweight receipt.
613    pub fn source(
614        &self,
615        diagnostic: BoundedDiagnostic,
616        classification: DiagnosticCode,
617        stage: RootRequestEvent,
618        axes: RootOutcomeFacts,
619    ) -> RootRequestFailure {
620        self.submit_source(
621            diagnostic.occurrence(),
622            diagnostic.category(),
623            SourceDetail::Bounded(&diagnostic),
624            classification,
625            stage,
626            axes,
627        )
628    }
629    /// Consume a previously captured source once. Never recaptures its occurrence,
630    /// location or stack; the dynamic legacy body is not retained in the receipt.
631    pub fn source_existing(
632        &self,
633        diagnostic: Diagnostic,
634        classification: DiagnosticCode,
635        stage: RootRequestEvent,
636        axes: RootOutcomeFacts,
637    ) -> RootRequestFailure {
638        self.submit_source(
639            diagnostic.occurrence(),
640            diagnostic.category(),
641            SourceDetail::Existing(&diagnostic),
642            classification,
643            stage,
644            axes,
645        )
646    }
647    fn submit_source(
648        &self,
649        occurrence: DiagnosticOccurrence,
650        category: DiagnosticCategory,
651        diagnostic: SourceDetail<'_>,
652        classification: DiagnosticCode,
653        stage: RootRequestEvent,
654        axes: RootOutcomeFacts,
655    ) -> RootRequestFailure {
656        let record = RootRecord {
657            schema_version: 2,
658            timestamp_unix_ms: timestamp(),
659            elapsed_ms: None,
660            level: if matches!(category, DiagnosticCategory::ExpectedRejection) {
661                "warn"
662            } else {
663                "error"
664            },
665            event: "request_failure_source",
666            stage,
667            context: self.view,
668            source_context: None,
669            occurrence: Some(occurrence),
670            classification: Some(classification),
671            source_submission: None,
672            original_capture: Some(OriginalCaptureState::LegacyProjectionOnly),
673            axes,
674            diagnostic: Some(diagnostic),
675            source_outcome: None,
676            category: Some(category),
677            detail_status: "bounded",
678        };
679        let fallback = RootRecord {
680            diagnostic: None,
681            detail_status: "omitted_encoding_capacity",
682            ..record
683        };
684        let submission = self
685            .output
686            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
687                output.submit_fixed_with_fallback(&record, &fallback)
688            });
689        RootRequestFailure {
690            source: self.view.clone(),
691            occurrence,
692            classification,
693            source_submission: submission,
694            category,
695            outcome: axes,
696            original_capture: OriginalCaptureState::LegacyProjectionOnly,
697        }
698    }
699    /// Ordinary stage/result events read exactly the same view, encoded before
700    /// queue admission. The regular logger receives only independent bytes.
701    pub fn ordinary(
702        &self,
703        observer: &Observer,
704        stage: RootRequestEvent,
705        axes: RootOutcomeFacts,
706    ) -> DiagnosticSubmission {
707        self.ordinary_at(observer, stage, axes, None, "request_stage")
708    }
709    fn ordinary_at(
710        &self,
711        observer: &Observer,
712        stage: RootRequestEvent,
713        axes: RootOutcomeFacts,
714        elapsed_ms: Option<u64>,
715        event: &'static str,
716    ) -> DiagnosticSubmission {
717        observer.emit_context_record(&crate::ContextRecord {
718            record_schema: "1.0", timestamp: timestamp(), level: "info", event,
719            context: self.view.unified_context(), payload: OrdinaryPayload { stage, elapsed_ms, axes },
720        })
721    }
722
723    /// Owns only an observation interval. No Call/Event copy, root allocation or
724    /// resource completion permission is hidden in this borrowed stage.
725    pub fn start_stage(
726        self,
727        observer: &'a Observer,
728        stage: RootRequestEvent,
729    ) -> RootActiveStage<'a> {
730        self.bounded_event(
731            stage,
732            RootOutcomeFacts::default(),
733            Some(0),
734            "request_stage_started",
735        );
736        self.ordinary_at(observer, stage, RootOutcomeFacts::default(), Some(0), "request_stage_started");
737        RootActiveStage {
738            scope: self,
739            observer,
740            stage,
741            started: std::time::Instant::now(),
742            finished: false,
743        }
744    }
745    /// A matched actual task join, after its supervisor removed the task slot.
746    /// It does not assert response delivery, database disposal or account release.
747    #[doc(hidden)]
748    pub fn task_reclaimed(&self, completion: &'static str, account: Option<saddle_admission::AccountIdentity>) -> DiagnosticSubmission {
749        #[derive(Serialize)]
750        struct Record<'a> {
751            schema: &'static str, event: &'static str, process_id: u32,
752            process_instance: u128, completion: &'static str,
753            context: &'a RequestExecutionView,
754            account: Option<saddle_admission::AccountIdentity>,
755        }
756        self.output.map_or(DiagnosticSubmission::OutputUnavailable, |output|
757            output.submit_fixed_record(&Record {
758                schema: "request-finalization/1", event: "request_task_reclaimed",
759                process_id: std::process::id(),
760                process_instance: crate::diagnostic::finalization_process_instance(),
761                completion, context: self.view, account,
762            }))
763    }
764
765    fn bounded_event(
766        &self,
767        stage: RootRequestEvent,
768        axes: RootOutcomeFacts,
769        elapsed_ms: Option<u64>,
770        event: &'static str,
771    ) -> DiagnosticSubmission {
772        self.output
773            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
774                output.submit_fixed_record(&RootRecord {
775                    schema_version: 2,
776                    timestamp_unix_ms: timestamp(),
777                    level: "info",
778                    elapsed_ms,
779                    event,
780                    stage,
781                    context: self.view,
782                    source_context: None,
783                    occurrence: None,
784                    classification: None,
785                    source_submission: None,
786                    original_capture: None,
787                    axes,
788                    source_outcome: None,
789                    category: None,
790                    detail_status: "not_applicable",
791                    diagnostic: None,
792                })
793            })
794    }
795}
796
797impl RootRequestFailure {
798    fn into_public(self, terminal_submission: DiagnosticSubmission) -> PublicRequestFailure {
799        PublicRequestFailure {
800            occurrence: self.occurrence,
801            classification: self.classification,
802            source_submission: self.source_submission,
803            terminal_submission,
804            category: self.category,
805            source_outcome: self.outcome,
806            original_capture: self.original_capture,
807        }
808    }
809    pub fn original_capture(&self) -> OriginalCaptureState {
810        self.original_capture
811    }
812    pub fn occurrence(&self) -> DiagnosticOccurrence {
813        self.occurrence
814    }
815    pub fn submission(&self) -> DiagnosticSubmission {
816        self.source_submission
817    }
818    pub fn source_view(&self) -> &RequestExecutionView {
819        &self.source
820    }
821    pub fn map_classification(mut self, classification: DiagnosticCode) -> Self {
822        self.classification = classification;
823        self
824    }
825    /// Foreign-root rejection returns the original receipt for a legitimate retry.
826    pub fn boundary(
827        self,
828        current: &RequestExecutionView,
829        output: Option<&EmergencyDiagnosticHandle>,
830        stage: RootRequestEvent,
831        axes: RootOutcomeFacts,
832    ) -> Result<(Self, DiagnosticSubmission), Self> {
833        self.boundary_at(current, output, stage, axes, None, "request_failure_boundary")
834    }
835    fn boundary_at(
836        self,
837        current: &RequestExecutionView,
838        output: Option<&EmergencyDiagnosticHandle>,
839        stage: RootRequestEvent,
840        axes: RootOutcomeFacts,
841        elapsed_ms: Option<u64>,
842        event: &'static str,
843    ) -> Result<(Self, DiagnosticSubmission), Self> {
844        if !self.source.same_request(current) {
845            return Err(self);
846        }
847        let submission = output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
848            output.submit_fixed_record(&RootRecord {
849                schema_version: 2,
850                timestamp_unix_ms: timestamp(),
851                level: "info",
852                elapsed_ms,
853                event,
854                stage,
855                context: current,
856                source_context: Some(&self.source),
857                occurrence: Some(self.occurrence),
858                classification: Some(self.classification),
859                source_submission: Some(self.source_submission),
860                original_capture: Some(self.original_capture),
861                axes,
862                diagnostic: None,
863                source_outcome: Some(self.outcome),
864                category: Some(self.category),
865                detail_status: "source_reference_only",
866            })
867        });
868        Ok((self, submission))
869    }
870    /// Submit final reference, then consume all receipt references. Caller must
871    /// release its other views/root before settling the original account.
872    pub fn finish(
873        self,
874        current: &RequestExecutionView,
875        output: Option<&EmergencyDiagnosticHandle>,
876        axes: RootOutcomeFacts,
877    ) -> Result<PublicRequestFailure, Self> {
878        let (receipt, terminal_submission) =
879            self.boundary(current, output, RootRequestEvent::Finalization, axes)?;
880        Ok(PublicRequestFailure {
881            occurrence: receipt.occurrence,
882            classification: receipt.classification,
883            source_submission: receipt.source_submission,
884            terminal_submission,
885            category: receipt.category,
886            source_outcome: receipt.outcome,
887            original_capture: receipt.original_capture,
888        })
889    }
890}
891
892/// Single terminal observation, preserving the existing fixed metric labels.
893/// Drop reports observation cancellation only, NOT physical cleanup or a source
894/// receipt. Runtime must separately handle a cancelled task's technical return.
895pub struct RootActiveStage<'a> {
896    scope: RootDiagnosticScope<'a>,
897    observer: &'a Observer,
898    stage: RootRequestEvent,
899    started: std::time::Instant,
900    finished: bool,
901}
902impl RootActiveStage<'_> {
903    fn finish_metrics(&mut self, outcome: saddle_core::OperationOutcome) -> u64 {
904        self.finished = true;
905        let elapsed = u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX);
906        use crate::{Stage, StageOutcome};
907        use saddle_core::OperationOutcome as Outcome;
908        let stage = match self.stage {
909            RootRequestEvent::Ingress => Some(Stage::Ingress),
910            RootRequestEvent::Admission => Some(Stage::Admission),
911            RootRequestEvent::Handler => Some(Stage::Handler),
912            RootRequestEvent::Database => Some(Stage::Database),
913            RootRequestEvent::Outbound => Some(Stage::ProfuseContract),
914            RootRequestEvent::Response => Some(Stage::Response),
915            RootRequestEvent::Finalization => Some(Stage::ResourceFinalization),
916            RootRequestEvent::Supervision => None,
917        };
918        let outcome = match outcome {
919            Outcome::Succeeded => Some(StageOutcome::Success),
920            Outcome::Rejected => Some(StageOutcome::Rejected),
921            Outcome::Failed | Outcome::Panicked => Some(StageOutcome::Failure),
922            Outcome::Cancelled | Outcome::TimedOut => Some(StageOutcome::Cancelled),
923            Outcome::Unknown => None,
924        };
925        if let (Some(stage), Some(outcome)) = (stage, outcome) {
926            self.observer
927                .inner
928                .metrics
929                .stage_finished(stage, outcome, elapsed);
930        }
931        elapsed
932    }
933    /// Only normal success/business rejection is accepted without a source.
934    pub fn finish_nonfailure(
935        mut self,
936        facts: RootOutcomeFacts,
937    ) -> Result<DiagnosticSubmission, Self> {
938        if !matches!(
939            facts.axes.operation,
940            saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
941        ) {
942            return Err(self);
943        }
944        let elapsed = self.finish_metrics(facts.axes.operation);
945        self.scope.ordinary_at(self.observer, self.stage, facts, Some(elapsed), "request_stage_finished");
946        Ok(self
947            .scope
948            .bounded_event(self.stage, facts, Some(elapsed), "request_stage_finished"))
949    }
950    /// Finish this interval and release its receipt without a second finalization boundary.
951    pub fn finish_failure_public(
952        self,
953        failure: RootRequestFailure,
954        facts: RootOutcomeFacts,
955    ) -> Result<PublicRequestFailure, (Self, RootRequestFailure)> {
956        self.finish_failure(failure, facts)
957            .map(|(failure, submission)| failure.into_public(submission))
958    }
959    /// Foreign failure returns both untouched owners, permitting original retry.
960    #[allow(clippy::result_large_err)] // Return owners without a rejection allocation.
961    pub fn finish_failure(
962        self,
963        failure: RootRequestFailure,
964        facts: RootOutcomeFacts,
965    ) -> Result<(RootRequestFailure, DiagnosticSubmission), (Self, RootRequestFailure)> {
966        self.finish_failure_record(failure, facts, "request_failure_boundary")
967    }
968    /// Finish metrics and the observed stage while the original receipt remains
969    /// task-owned. Only its later consuming projection is the required boundary.
970    #[allow(clippy::result_large_err)] // Preserve both owners without allocating.
971    pub fn finish_failure_retained(
972        self,
973        failure: RootRequestFailure,
974        facts: RootOutcomeFacts,
975    ) -> Result<(RootRequestFailure, DiagnosticSubmission), (Self, RootRequestFailure)> {
976        self.finish_failure_record(failure, facts, "request_stage_finished")
977    }
978    #[allow(clippy::result_large_err)] // Preserve both owners without allocating.
979    fn finish_failure_record(
980        mut self,
981        failure: RootRequestFailure,
982        facts: RootOutcomeFacts,
983        event: &'static str,
984    ) -> Result<(RootRequestFailure, DiagnosticSubmission), (Self, RootRequestFailure)> {
985        if !failure.source.same_request(self.scope.view) {
986            return Err((self, failure));
987        }
988        let elapsed = self.finish_metrics(facts.axes.operation);
989        match failure.boundary_at(
990            self.scope.view,
991            self.scope.output,
992            self.stage,
993            facts,
994            Some(elapsed),
995            event,
996        ) {
997            Ok(result) => {
998                self.scope.ordinary_at(self.observer, self.stage, facts, Some(elapsed), event);
999                Ok(result)
1000            },
1001            Err(failure) => Err((self, failure)),
1002        }
1003    }
1004}
1005impl Drop for RootActiveStage<'_> {
1006    fn drop(&mut self) {
1007        if !self.finished {
1008            let elapsed = self.finish_metrics(saddle_core::OperationOutcome::Cancelled);
1009            let facts = RootOutcomeFacts {
1010                axes: DiagnosticOutcomeAxes {
1011                    operation: saddle_core::OperationOutcome::Cancelled,
1012                    ..Default::default()
1013                },
1014                ..Default::default()
1015            };
1016            self.scope.ordinary_at(self.observer, self.stage, facts, Some(elapsed), "request_stage_finished");
1017            self.scope
1018                .bounded_event(self.stage, facts, Some(elapsed), "request_stage_finished");
1019        }
1020    }
1021}
1022
1023pub fn root_failure_layout() -> std::alloc::Layout {
1024    std::alloc::Layout::new::<RootRequestFailure>()
1025}
1026
1027/// Concrete payload layouts for R0, not allocator charge or reservation authority.
1028/// Channel internals and active sender/worker ownership must also be included by
1029/// the process/log-domain storage contract; slot count alone is not that proof.
1030pub struct RequestLoggingLayouts {
1031    pub source_frame: std::alloc::Layout,
1032    pub emergency_packet: std::alloc::Layout,
1033    pub emergency_slots: usize,
1034    pub ordinary_command: std::alloc::Layout,
1035    pub ordinary_encoded_bytes_max: usize,
1036    pub source_record: std::alloc::Layout,
1037    /// One process-owned Arc allocation: counters, confirmation and bounded takeover target.
1038    pub confirmation_owner: std::alloc::Layout,
1039}
1040pub fn request_logging_layouts() -> RequestLoggingLayouts {
1041    let (source_frame, emergency_packet, emergency_slots) = crate::diagnostic::root_frame_layouts();
1042    RequestLoggingLayouts {
1043        source_frame,
1044        emergency_packet,
1045        emergency_slots,
1046        ordinary_command: crate::logger::root_queue_layout(),
1047        ordinary_encoded_bytes_max: 8191,
1048        source_record: std::alloc::Layout::new::<RootRecord<'static>>(),
1049        confirmation_owner: crate::diagnostic::root_status_owner_layout(),
1050    }
1051}
1052
1053impl PublicRequestFailure {
1054    pub fn occurrence(&self) -> DiagnosticOccurrence {
1055        self.occurrence
1056    }
1057    pub fn source_submission(&self) -> DiagnosticSubmission {
1058        self.source_submission
1059    }
1060    pub fn terminal_submission(&self) -> DiagnosticSubmission {
1061        self.terminal_submission
1062    }
1063}
1064impl std::fmt::Debug for PublicRequestFailure {
1065    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1066        f.debug_struct("PublicRequestFailure")
1067            .field("classification", &self.classification)
1068            .field("source_submission", &self.source_submission)
1069            .finish_non_exhaustive()
1070    }
1071}
1072impl std::fmt::Display for PublicRequestFailure {
1073    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1074        write!(f, "framework request failure: {:?}", self.classification)
1075    }
1076}
1077impl std::error::Error for PublicRequestFailure {}
1078
1079/// Completed maintenance response only; not a request or physical-return claim.
1080#[doc(hidden)]
1081pub fn database_maintenance_success(
1082    output: Option<&EmergencyDiagnosticHandle>, datasource: u64, connection: u64,
1083    sweep: u64, recovery_acquisition: bool,
1084) -> DiagnosticSubmission {
1085    #[derive(Serialize)]
1086    struct CompletedProbe {
1087        schema_version: u8,
1088        event: &'static str,
1089        datasource: u64,
1090        connection: u64,
1091        maintenance_sweep: u64,
1092        recovery_acquisition: bool,
1093        response: &'static str,
1094        request: ContextFact<()>,
1095        trace_id: ContextFact<()>,
1096    }
1097    match output {
1098        Some(output) => output.submit_fixed_record(&CompletedProbe {
1099            schema_version: 1, event: "database_maintenance_probe", datasource, connection,
1100            maintenance_sweep: sweep, recovery_acquisition, response: "complete_valid_select1",
1101            request: ContextFact::NotApplicable, trace_id: ContextFact::NotApplicable,
1102        }),
1103        None => DiagnosticSubmission::OutputUnavailable,
1104    }
1105}
1106
1107pub use original::{request_terminal_audit_error_owned, request_terminal_audit_disposition_owned,
1108    request_terminal_audit_passed_owned, request_terminal_audit_cleanup_owned};
saddle_observability/root_diagnostic.rs

saddle_observability/
root_diagnostic.rs
