saddle_observability/
fifo.rs

1//! Read-only transitions at the formal ingress queue operations.
2use crate::{DiagnosticSubmission,EmergencyDiagnosticHandle,Observer};
3use saddle_core::request_context::ContextIdentity;
4use serde::Serialize;
5
6impl Observer {
7    /// Original same-lock ledger sample after application driver join and
8    /// physical pool reclaim. Values are not manufactured from join success.
9    pub fn record_standard_rpc_shutdown(&self, output: Option<&EmergencyDiagnosticHandle>,
10        joined: bool, rpc: &saddle_admission::RpcStageSnapshot,
11    ) -> DiagnosticSubmission {
12        #[derive(Serialize)]
13        struct Record {
14            schema: &'static str, event: &'static str, source: &'static str,
15            process_id: u32, process_generation: u128, driver_joined: bool,
16            pool_physical_owners: usize, pool_storage_bytes: usize,
17            request_physical_references: usize, request_backing_bytes: usize,
18        }
19        let Some(output) = output else { return DiagnosticSubmission::OutputUnavailable; };
20        output.submit_fixed_record(&Record {
21            schema: "standard-rpc-shutdown/1", event: "framework.standard_rpc_shutdown",
22            source: "post-driver-join-and-pool-reclaim-original-ledger",
23            process_id: std::process::id(), process_generation: crate::diagnostic::finalization_process_instance(),
24            driver_joined: joined, pool_physical_owners: rpc.standard_pool_owners,
25            pool_storage_bytes: rpc.standard_pool_storage_bytes,
26            request_physical_references: rpc.physical_references, request_backing_bytes: rpc.backing_bytes,
27        })
28    }
29}
30
31impl Observer {
32    pub fn record_ingress_fifo(&self,output:Option<&EmergencyDiagnosticHandle>,request_id:Option<ContextIdentity>,
33        queue_sequence:u64,transition:&'static str,reason:&'static str)->DiagnosticSubmission {
34        #[derive(Serialize)]
35        struct Record {
36            schema:&'static str,event:&'static str,process_id:u32,process_generation:u128,
37            request_id:Option<ContextIdentity>,queue_sequence:u64,transition:&'static str,reason:&'static str,
38        }
39        let Some(output)=output else{return DiagnosticSubmission::OutputUnavailable;};
40        output.submit_fixed_record(&Record {schema:"ingress-fifo/1",event:"framework.ingress_fifo",
41            process_id:std::process::id(),process_generation:crate::diagnostic::finalization_process_instance(),
42            request_id,queue_sequence,transition,reason})
43    }
44}
45
46impl Observer {
47    /// Attempt identity only. Resource amounts remain on the original same-lock
48    /// capacity receipt; input_peak is the already recognized input demand.
49    pub fn record_ingress_fifo_attempt(&self, output: Option<&EmergencyDiagnosticHandle>,
50        request_id: Option<ContextIdentity>, queue_sequence: u64, decision_attempt: u64,
51        input_peak: usize, outcome: &'static str) -> DiagnosticSubmission {
52        #[derive(Serialize)]
53        struct Record {
54            schema: &'static str, event: &'static str, process_id: u32,
55            process_generation: u128, request_id: Option<ContextIdentity>,
56            queue_sequence: u64, decision_attempt: u64, input_peak: usize, outcome: &'static str,
57        }
58        let Some(output) = output else { return DiagnosticSubmission::OutputUnavailable; };
59        output.submit_fixed_record(&Record {schema: "ingress-fifo-attempt/1",
60            event: "framework.ingress_fifo_attempt", process_id: std::process::id(),
61            process_generation: crate::diagnostic::finalization_process_instance(),
62            request_id, queue_sequence, decision_attempt, input_peak, outcome})
63    }
64}
65
66/// The capacity receipt and FIFO marker use this same diagnostic process epoch.
67#[doc(hidden)]
68pub fn ingress_fifo_process_generation() -> u128 {
69    crate::diagnostic::finalization_process_instance()
70}
71
72impl Observer {
73    /// A post-join sample, not an admission decision or an atomic combined ledger receipt.
74    pub fn record_resource_quiescent(&self, output: Option<&EmergencyDiagnosticHandle>,
75        context: Option<saddle_core::request_context::RequestContextSnapshot>, completion_sequence: u64,
76        resource: &saddle_admission::ProcessSnapshot, rpc: &saddle_admission::RpcStageSnapshot,
77    ) -> DiagnosticSubmission {
78        #[derive(Serialize)]
79        struct Record {
80            schema: &'static str, event: &'static str, source: &'static str,
81            process_id: u32, process_generation: u128,
82            context: Option<saddle_core::request_context::RequestContextSnapshot>, completion_sequence: u64,
83            active_accounts: usize, charged: usize, task_charged: usize, framework_charged: usize,
84            registered_waiters: usize, healthy: bool, rpc_capacity: usize,
85            rpc_operations: usize, rpc_physical_references: usize,
86        }
87        let Some(output) = output else { return DiagnosticSubmission::OutputUnavailable; };
88        output.submit_fixed_record(&Record {schema: "resource-quiescent/1", event: "framework.resource_quiescent",
89            source: "postjoin-empty-collections-original-ledger-samples", process_id: std::process::id(),
90            process_generation: crate::diagnostic::finalization_process_instance(), context, completion_sequence,
91            active_accounts: resource.active_accounts, charged: resource.charged,
92            task_charged: resource.task_charged, framework_charged: resource.framework_charged,
93            registered_waiters: resource.registered_waiters, healthy: resource.healthy,
94            rpc_capacity: rpc.capacity, rpc_operations: rpc.operations, rpc_physical_references: rpc.physical_references})
95    }
96}
97
98impl EmergencyDiagnosticHandle {
99    /// Scalar recording-point evidence after the caller's original queue mutation.
100    /// This is not an exact physical removal timestamp or a new deadline owner.
101    pub fn record_ingress_fifo_clock(&self, request_id: Option<ContextIdentity>,
102        queue_sequence: u64, transition: &'static str, reason: &'static str,
103        effective_deadline_unix_ms: i64) -> DiagnosticSubmission {
104        #[derive(Serialize)]
105        struct Record {
106            schema: &'static str, event: &'static str, process_id: u32,
107            process_generation: u128, request_id: Option<ContextIdentity>,
108            queue_sequence: u64, transition: &'static str, reason: &'static str,
109            effective_deadline_unix_ms: i64, recorded_unix_ns: Option<u128>,
110        }
111        let recorded_unix_ns = std::time::SystemTime::now()
112            .duration_since(std::time::UNIX_EPOCH).ok().map(|elapsed| elapsed.as_nanos());
113        self.submit_fixed_record(&Record {
114            schema: "ingress-fifo-clock/1", event: "framework.ingress_fifo_clock",
115            process_id: std::process::id(),
116            process_generation: crate::diagnostic::finalization_process_instance(),
117            request_id, queue_sequence, transition, reason,
118            effective_deadline_unix_ms, recorded_unix_ns,
119        })
120    }
121}
saddle_observability/fifo.rs

saddle_observability/
fifo.rs
