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;
8mod original;
9pub use original::{
10    OriginalCaptureState, UnrootedCaptureFacts, UnrootedDiagnosticScope, original_capture_layout,
11    database_background_error,
12};
13
14/// Read the existing typed call/event; this does not create a second context.
15pub fn request_identity_group(
16    call: &saddle_core::CallContext,
17    event: &EventContext,
18    zone: ContextFact<saddle_core::ContextLabel>,
19) -> Result<saddle_core::RequestIdentityGroup, saddle_core::ContextConflict> {
20    saddle_core::RequestIdentityGroup::from_validated(
21        call,
22        event.diagnostic_request(),
23        event.diagnostic_route(),
24        event.diagnostic_attempt(),
25        zone,
26    )
27}
28/// Consume the exact child Call/Event already established by Boundary.
29pub fn request_child_view(
30    parent: &RequestExecutionView,
31    call: &saddle_core::CallContext,
32    event: &EventContext,
33) -> Result<RequestExecutionView, saddle_core::ContextConflict> {
34    parent.child(
35        call,
36        event.diagnostic_request(),
37        event.diagnostic_route(),
38        event.diagnostic_attempt(),
39    )
40}
41
42/// Closed event families, never a caller-supplied JSON key or arbitrary message.
43#[derive(Clone, Copy, Debug, Serialize)]
44#[serde(rename_all = "snake_case")]
45pub enum RootRequestEvent {
46    Ingress,
47    Admission,
48    Handler,
49    Database,
50    Outbound,
51    Response,
52    Finalization,
53    Supervision,
54}
55
56/// Reported only by the DB owner. Rejected is NOT proof of confirmed rollback.
57#[derive(Clone, Copy, Serialize)]
58#[serde(rename_all = "snake_case")]
59pub enum RequestTransactionFact {
60    Committed,
61    Rejected,
62    Unknown,
63}
64#[derive(Clone, Copy, Serialize)]
65pub struct RootOutcomeFacts {
66    pub axes: DiagnosticOutcomeAxes,
67    pub transaction: ContextFact<RequestTransactionFact>,
68}
69impl Default for RootOutcomeFacts {
70    fn default() -> Self {
71        Self {
72            axes: DiagnosticOutcomeAxes::default(),
73            transaction: ContextFact::Unavailable,
74        }
75    }
76}
77
78#[derive(Serialize)]
79#[serde(untagged)]
80enum SourceDetail<'a> {
81    Bounded(&'a BoundedDiagnostic),
82    Existing(&'a Diagnostic),
83}
84
85fn timestamp() -> u128 {
86    std::time::SystemTime::now()
87        .duration_since(std::time::UNIX_EPOCH)
88        .unwrap_or_default()
89        .as_millis()
90}
91
92#[derive(Serialize)]
93struct RootRecord<'a> {
94    schema_version: u8,
95    timestamp_unix_ms: u128,
96    level: &'static str,
97    elapsed_ms: Option<u64>,
98    event: &'static str,
99    stage: RootRequestEvent,
100    context: &'a RequestExecutionView,
101    source_context: Option<&'a RequestExecutionView>,
102    occurrence: Option<DiagnosticOccurrence>,
103    classification: Option<DiagnosticCode>,
104    source_submission: Option<DiagnosticSubmission>,
105    original_capture: Option<OriginalCaptureState>,
106    axes: RootOutcomeFacts,
107    source_outcome: Option<RootOutcomeFacts>,
108    category: Option<DiagnosticCategory>,
109    detail_status: &'static str,
110    diagnostic: Option<SourceDetail<'a>>,
111}
112
113/// Light, single-consumption source receipt: no exception body, original error,
114/// copied root facts, output handle, task/DB capability or account owner.
115///
116/// ```compile_fail
117/// use saddle_observability::RootRequestFailure;
118/// fn duplicate(f: RootRequestFailure) { let _ = f.clone(); }
119/// ```
120/// ```compile_fail
121/// use saddle_observability::RootRequestFailure;
122/// let f = RootRequestFailure {};
123/// ```
124#[must_use = "carry the source receipt to its declared terminal boundary"]
125pub struct RootRequestFailure {
126    source: RequestExecutionView,
127    occurrence: DiagnosticOccurrence,
128    classification: DiagnosticCode,
129    source_submission: DiagnosticSubmission,
130    category: DiagnosticCategory,
131    outcome: RootOutcomeFacts,
132    original_capture: OriginalCaptureState,
133}
134
135/// Root-free public error projection. Holding it cannot keep any request alive.
136#[derive(Clone, Copy, Serialize)]
137pub struct PublicRequestFailure {
138    occurrence: DiagnosticOccurrence,
139    classification: DiagnosticCode,
140    source_submission: DiagnosticSubmission,
141    terminal_submission: DiagnosticSubmission,
142    category: DiagnosticCategory,
143    source_outcome: RootOutcomeFacts,
144    original_capture: OriginalCaptureState,
145}
146
147/// A supervision result cannot be a naked technical code: failure consumes the
148/// original source receipt. Runtime retains its own physical ownership outside it.
149///
150/// ```compile_fail
151/// use saddle_observability::RootSupervisionReturn;
152/// use saddle_core::DiagnosticCode;
153/// let _: RootSupervisionReturn<()> = RootSupervisionReturn::failed(DiagnosticCode::new("task.failed").unwrap());
154/// ```
155/// ```compile_fail
156/// use saddle_observability::{RootSupervisionReturn, RootRequestFailure};
157/// fn replay(f: RootRequestFailure) {
158///     let _first = RootSupervisionReturn::<()>::failed(f);
159///     let _second = RootSupervisionReturn::<()>::failed(f);
160/// }
161/// ```
162#[must_use = "the supervisor must consume the actual task result"]
163pub struct RootSupervisionReturn<T> {
164    result: Result<T, RootRequestFailure>,
165}
166impl<T> RootSupervisionReturn<T> {
167    pub fn completed(value: T) -> Self {
168        Self { result: Ok(value) }
169    }
170    pub fn failed(failure: RootRequestFailure) -> Self {
171        Self {
172            result: Err(failure),
173        }
174    }
175    pub fn consume(self) -> Result<T, RootRequestFailure> {
176        self.result
177    }
178}
179
180/// Borrowed logging scope. No output handle or logger ever enters the root.
181pub struct RootDiagnosticScope<'a> {
182    view: &'a RequestExecutionView,
183    output: Option<&'a EmergencyDiagnosticHandle>,
184}
185impl<'a> RootDiagnosticScope<'a> {
186    pub fn new(
187        view: &'a RequestExecutionView,
188        output: Option<&'a EmergencyDiagnosticHandle>,
189    ) -> Self {
190        Self { view, output }
191    }
192    /// Detail is consumed and dropped after the one bounded output attempt. Even
193    /// OutputUnavailable/Full/EncodingFailed returns the same lightweight receipt.
194    pub fn source(
195        &self,
196        diagnostic: BoundedDiagnostic,
197        classification: DiagnosticCode,
198        stage: RootRequestEvent,
199        axes: RootOutcomeFacts,
200    ) -> RootRequestFailure {
201        self.submit_source(
202            diagnostic.occurrence(),
203            diagnostic.category(),
204            SourceDetail::Bounded(&diagnostic),
205            classification,
206            stage,
207            axes,
208        )
209    }
210    /// Consume a previously captured source once. Never recaptures its occurrence,
211    /// location or stack; the dynamic legacy body is not retained in the receipt.
212    pub fn source_existing(
213        &self,
214        diagnostic: Diagnostic,
215        classification: DiagnosticCode,
216        stage: RootRequestEvent,
217        axes: RootOutcomeFacts,
218    ) -> RootRequestFailure {
219        self.submit_source(
220            diagnostic.occurrence(),
221            diagnostic.category(),
222            SourceDetail::Existing(&diagnostic),
223            classification,
224            stage,
225            axes,
226        )
227    }
228    fn submit_source(
229        &self,
230        occurrence: DiagnosticOccurrence,
231        category: DiagnosticCategory,
232        diagnostic: SourceDetail<'_>,
233        classification: DiagnosticCode,
234        stage: RootRequestEvent,
235        axes: RootOutcomeFacts,
236    ) -> RootRequestFailure {
237        let record = RootRecord {
238            schema_version: 2,
239            timestamp_unix_ms: timestamp(),
240            elapsed_ms: None,
241            level: if matches!(category, DiagnosticCategory::ExpectedRejection) {
242                "warn"
243            } else {
244                "error"
245            },
246            event: "request_failure_source",
247            stage,
248            context: self.view,
249            source_context: None,
250            occurrence: Some(occurrence),
251            classification: Some(classification),
252            source_submission: None,
253            original_capture: Some(OriginalCaptureState::LegacyProjectionOnly),
254            axes,
255            diagnostic: Some(diagnostic),
256            source_outcome: None,
257            category: Some(category),
258            detail_status: "bounded",
259        };
260        let fallback = RootRecord {
261            diagnostic: None,
262            detail_status: "omitted_encoding_capacity",
263            ..record
264        };
265        let submission = self
266            .output
267            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
268                output.submit_fixed_with_fallback(&record, &fallback)
269            });
270        RootRequestFailure {
271            source: self.view.clone(),
272            occurrence,
273            classification,
274            source_submission: submission,
275            category,
276            outcome: axes,
277            original_capture: OriginalCaptureState::LegacyProjectionOnly,
278        }
279    }
280    /// Ordinary stage/result events read exactly the same view, encoded before
281    /// queue admission. The regular logger receives only independent bytes.
282    pub fn ordinary(
283        &self,
284        observer: &Observer,
285        stage: RootRequestEvent,
286        axes: RootOutcomeFacts,
287    ) -> DiagnosticSubmission {
288        self.ordinary_at(observer, stage, axes, None, "request_stage")
289    }
290    fn ordinary_at(
291        &self,
292        observer: &Observer,
293        stage: RootRequestEvent,
294        axes: RootOutcomeFacts,
295        elapsed_ms: Option<u64>,
296        event: &'static str,
297    ) -> DiagnosticSubmission {
298        observer.emit_root_record(&RootRecord {
299            schema_version: 2,
300            timestamp_unix_ms: timestamp(),
301            level: "info",
302            elapsed_ms,
303            event,
304            stage,
305            context: self.view,
306            source_context: None,
307            occurrence: None,
308            classification: None,
309            source_submission: None,
310            original_capture: None,
311            axes,
312            diagnostic: None,
313            source_outcome: None,
314            category: None,
315            detail_status: "not_applicable",
316        })
317    }
318    /// Owns only an observation interval. No Call/Event copy, root allocation or
319    /// resource completion permission is hidden in this borrowed stage.
320    pub fn start_stage(
321        self,
322        observer: &'a Observer,
323        stage: RootRequestEvent,
324    ) -> RootActiveStage<'a> {
325        self.bounded_event(
326            stage,
327            RootOutcomeFacts::default(),
328            Some(0),
329            "request_stage_started",
330        );
331        RootActiveStage {
332            scope: self,
333            observer,
334            stage,
335            started: std::time::Instant::now(),
336            finished: false,
337        }
338    }
339    fn bounded_event(
340        &self,
341        stage: RootRequestEvent,
342        axes: RootOutcomeFacts,
343        elapsed_ms: Option<u64>,
344        event: &'static str,
345    ) -> DiagnosticSubmission {
346        self.output
347            .map_or(DiagnosticSubmission::OutputUnavailable, |output| {
348                output.submit_fixed_record(&RootRecord {
349                    schema_version: 2,
350                    timestamp_unix_ms: timestamp(),
351                    level: "info",
352                    elapsed_ms,
353                    event,
354                    stage,
355                    context: self.view,
356                    source_context: None,
357                    occurrence: None,
358                    classification: None,
359                    source_submission: None,
360                    original_capture: None,
361                    axes,
362                    source_outcome: None,
363                    category: None,
364                    detail_status: "not_applicable",
365                    diagnostic: None,
366                })
367            })
368    }
369}
370
371impl RootRequestFailure {
372    fn into_public(self, terminal_submission: DiagnosticSubmission) -> PublicRequestFailure {
373        PublicRequestFailure {
374            occurrence: self.occurrence,
375            classification: self.classification,
376            source_submission: self.source_submission,
377            terminal_submission,
378            category: self.category,
379            source_outcome: self.outcome,
380            original_capture: self.original_capture,
381        }
382    }
383    pub fn original_capture(&self) -> OriginalCaptureState {
384        self.original_capture
385    }
386    pub fn occurrence(&self) -> DiagnosticOccurrence {
387        self.occurrence
388    }
389    pub fn submission(&self) -> DiagnosticSubmission {
390        self.source_submission
391    }
392    pub fn source_view(&self) -> &RequestExecutionView {
393        &self.source
394    }
395    pub fn map_classification(mut self, classification: DiagnosticCode) -> Self {
396        self.classification = classification;
397        self
398    }
399    /// Foreign-root rejection returns the original receipt for a legitimate retry.
400    pub fn boundary(
401        self,
402        current: &RequestExecutionView,
403        output: Option<&EmergencyDiagnosticHandle>,
404        stage: RootRequestEvent,
405        axes: RootOutcomeFacts,
406    ) -> Result<(Self, DiagnosticSubmission), Self> {
407        self.boundary_at(current, output, stage, axes, None, "request_failure_boundary")
408    }
409    fn boundary_at(
410        self,
411        current: &RequestExecutionView,
412        output: Option<&EmergencyDiagnosticHandle>,
413        stage: RootRequestEvent,
414        axes: RootOutcomeFacts,
415        elapsed_ms: Option<u64>,
416        event: &'static str,
417    ) -> Result<(Self, DiagnosticSubmission), Self> {
418        if !self.source.same_request(current) {
419            return Err(self);
420        }
421        let submission = output.map_or(DiagnosticSubmission::OutputUnavailable, |output| {
422            output.submit_fixed_record(&RootRecord {
423                schema_version: 2,
424                timestamp_unix_ms: timestamp(),
425                level: "info",
426                elapsed_ms,
427                event,
428                stage,
429                context: current,
430                source_context: Some(&self.source),
431                occurrence: Some(self.occurrence),
432                classification: Some(self.classification),
433                source_submission: Some(self.source_submission),
434                original_capture: Some(self.original_capture),
435                axes,
436                diagnostic: None,
437                source_outcome: Some(self.outcome),
438                category: Some(self.category),
439                detail_status: "source_reference_only",
440            })
441        });
442        Ok((self, submission))
443    }
444    /// Submit final reference, then consume all receipt references. Caller must
445    /// release its other views/root before settling the original account.
446    pub fn finish(
447        self,
448        current: &RequestExecutionView,
449        output: Option<&EmergencyDiagnosticHandle>,
450        axes: RootOutcomeFacts,
451    ) -> Result<PublicRequestFailure, Self> {
452        let (receipt, terminal_submission) =
453            self.boundary(current, output, RootRequestEvent::Finalization, axes)?;
454        Ok(PublicRequestFailure {
455            occurrence: receipt.occurrence,
456            classification: receipt.classification,
457            source_submission: receipt.source_submission,
458            terminal_submission,
459            category: receipt.category,
460            source_outcome: receipt.outcome,
461            original_capture: receipt.original_capture,
462        })
463    }
464}
465
466/// Single terminal observation, preserving the existing fixed metric labels.
467/// Drop reports observation cancellation only, NOT physical cleanup or a source
468/// receipt. Runtime must separately handle a cancelled task's technical return.
469pub struct RootActiveStage<'a> {
470    scope: RootDiagnosticScope<'a>,
471    observer: &'a Observer,
472    stage: RootRequestEvent,
473    started: std::time::Instant,
474    finished: bool,
475}
476impl RootActiveStage<'_> {
477    fn finish_metrics(&mut self, outcome: saddle_core::OperationOutcome) -> u64 {
478        self.finished = true;
479        let elapsed = u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX);
480        use crate::{Stage, StageOutcome};
481        use saddle_core::OperationOutcome as Outcome;
482        let stage = match self.stage {
483            RootRequestEvent::Ingress => Some(Stage::Ingress),
484            RootRequestEvent::Admission => Some(Stage::Admission),
485            RootRequestEvent::Handler => Some(Stage::Handler),
486            RootRequestEvent::Database => Some(Stage::Database),
487            RootRequestEvent::Outbound => Some(Stage::ProfuseContract),
488            RootRequestEvent::Response => Some(Stage::Response),
489            RootRequestEvent::Finalization => Some(Stage::ResourceFinalization),
490            RootRequestEvent::Supervision => None,
491        };
492        let outcome = match outcome {
493            Outcome::Succeeded => Some(StageOutcome::Success),
494            Outcome::Rejected => Some(StageOutcome::Rejected),
495            Outcome::Failed | Outcome::Panicked => Some(StageOutcome::Failure),
496            Outcome::Cancelled | Outcome::TimedOut => Some(StageOutcome::Cancelled),
497            Outcome::Unknown => None,
498        };
499        if let (Some(stage), Some(outcome)) = (stage, outcome) {
500            self.observer
501                .inner
502                .metrics
503                .stage_finished(stage, outcome, elapsed);
504        }
505        elapsed
506    }
507    /// Only normal success/business rejection is accepted without a source.
508    pub fn finish_nonfailure(
509        mut self,
510        facts: RootOutcomeFacts,
511    ) -> Result<DiagnosticSubmission, Self> {
512        if !matches!(
513            facts.axes.operation,
514            saddle_core::OperationOutcome::Succeeded | saddle_core::OperationOutcome::Rejected
515        ) {
516            return Err(self);
517        }
518        let elapsed = self.finish_metrics(facts.axes.operation);
519        Ok(self
520            .scope
521            .bounded_event(self.stage, facts, Some(elapsed), "request_stage_finished"))
522    }
523    /// Finish this interval and release its receipt without a second finalization boundary.
524    pub fn finish_failure_public(
525        self,
526        failure: RootRequestFailure,
527        facts: RootOutcomeFacts,
528    ) -> Result<PublicRequestFailure, (Self, RootRequestFailure)> {
529        self.finish_failure(failure, facts)
530            .map(|(failure, submission)| failure.into_public(submission))
531    }
532    /// Foreign failure returns both untouched owners, permitting original retry.
533    #[allow(clippy::result_large_err)] // Return owners without a rejection allocation.
534    pub fn finish_failure(
535        self,
536        failure: RootRequestFailure,
537        facts: RootOutcomeFacts,
538    ) -> Result<(RootRequestFailure, DiagnosticSubmission), (Self, RootRequestFailure)> {
539        self.finish_failure_record(failure, facts, "request_failure_boundary")
540    }
541    /// Finish metrics and the observed stage while the original receipt remains
542    /// task-owned. Only its later consuming projection is the required boundary.
543    #[allow(clippy::result_large_err)] // Preserve both owners without allocating.
544    pub fn finish_failure_retained(
545        self,
546        failure: RootRequestFailure,
547        facts: RootOutcomeFacts,
548    ) -> Result<(RootRequestFailure, DiagnosticSubmission), (Self, RootRequestFailure)> {
549        self.finish_failure_record(failure, facts, "request_stage_finished")
550    }
551    #[allow(clippy::result_large_err)] // Preserve both owners without allocating.
552    fn finish_failure_record(
553        mut self,
554        failure: RootRequestFailure,
555        facts: RootOutcomeFacts,
556        event: &'static str,
557    ) -> Result<(RootRequestFailure, DiagnosticSubmission), (Self, RootRequestFailure)> {
558        if !failure.source.same_request(self.scope.view) {
559            return Err((self, failure));
560        }
561        let elapsed = self.finish_metrics(facts.axes.operation);
562        match failure.boundary_at(
563            self.scope.view,
564            self.scope.output,
565            self.stage,
566            facts,
567            Some(elapsed),
568            event,
569        ) {
570            Ok(result) => Ok(result),
571            Err(failure) => Err((self, failure)),
572        }
573    }
574}
575impl Drop for RootActiveStage<'_> {
576    fn drop(&mut self) {
577        if !self.finished {
578            let elapsed = self.finish_metrics(saddle_core::OperationOutcome::Cancelled);
579            let facts = RootOutcomeFacts {
580                axes: DiagnosticOutcomeAxes {
581                    operation: saddle_core::OperationOutcome::Cancelled,
582                    ..Default::default()
583                },
584                ..Default::default()
585            };
586            self.scope
587                .bounded_event(self.stage, facts, Some(elapsed), "request_stage_finished");
588        }
589    }
590}
591
592pub fn root_failure_layout() -> std::alloc::Layout {
593    std::alloc::Layout::new::<RootRequestFailure>()
594}
595
596/// Concrete payload layouts for R0, not allocator charge or reservation authority.
597/// Channel internals and active sender/worker ownership must also be included by
598/// the process/log-domain storage contract; slot count alone is not that proof.
599pub struct RequestLoggingLayouts {
600    pub source_frame: std::alloc::Layout,
601    pub emergency_packet: std::alloc::Layout,
602    pub emergency_slots: usize,
603    pub ordinary_command: std::alloc::Layout,
604    pub ordinary_encoded_bytes_max: usize,
605    pub source_record: std::alloc::Layout,
606}
607pub fn request_logging_layouts() -> RequestLoggingLayouts {
608    let (source_frame, emergency_packet, emergency_slots) = crate::diagnostic::root_frame_layouts();
609    RequestLoggingLayouts {
610        source_frame,
611        emergency_packet,
612        emergency_slots,
613        ordinary_command: crate::logger::root_queue_layout(),
614        ordinary_encoded_bytes_max: 8191,
615        source_record: std::alloc::Layout::new::<RootRecord<'static>>(),
616    }
617}
618
619impl PublicRequestFailure {
620    pub fn occurrence(&self) -> DiagnosticOccurrence {
621        self.occurrence
622    }
623    pub fn source_submission(&self) -> DiagnosticSubmission {
624        self.source_submission
625    }
626    pub fn terminal_submission(&self) -> DiagnosticSubmission {
627        self.terminal_submission
628    }
629}
630impl std::fmt::Debug for PublicRequestFailure {
631    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
632        f.debug_struct("PublicRequestFailure")
633            .field("classification", &self.classification)
634            .field("source_submission", &self.source_submission)
635            .finish_non_exhaustive()
636    }
637}
638impl std::fmt::Display for PublicRequestFailure {
639    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
640        write!(f, "framework request failure: {:?}", self.classification)
641    }
642}
643impl std::error::Error for PublicRequestFailure {}
644
645/// Completed maintenance response only; not a request or physical-return claim.
646#[doc(hidden)]
647pub fn database_maintenance_success(
648    output: Option<&EmergencyDiagnosticHandle>, datasource: u64, connection: u64,
649    sweep: u64, recovery_acquisition: bool,
650) -> DiagnosticSubmission {
651    #[derive(Serialize)]
652    struct CompletedProbe {
653        schema_version: u8,
654        event: &'static str,
655        datasource: u64,
656        connection: u64,
657        maintenance_sweep: u64,
658        recovery_acquisition: bool,
659        response: &'static str,
660        request: ContextFact<()>,
661        trace_id: ContextFact<()>,
662    }
663    match output {
664        Some(output) => output.submit_fixed_record(&CompletedProbe {
665            schema_version: 1, event: "database_maintenance_probe", datasource, connection,
666            maintenance_sweep: sweep, recovery_acquisition, response: "complete_valid_select1",
667            request: ContextFact::NotApplicable, trace_id: ContextFact::NotApplicable,
668        }),
669        None => DiagnosticSubmission::OutputUnavailable,
670    }
671}
saddle_observability/root_diagnostic.rs

saddle_observability/
root_diagnostic.rs
