saddle_observability/
admission.rs

1//! Authority-free output of an existing admission decision. Runtime owns and
2//! consumes the original receipt; this module neither admits nor creates roots.
3use crate::{CapacityDimension, DiagnosticSubmission, EmergencyDiagnosticHandle, Observer};
4use saddle_admission::{
5    AdmissionError, ProfuseGwAdmissionObservationReceipt, ProfuseGwCapacityBottleneck, ProfuseGwCapacityDecision,
6    ProfuseGwCapacityRejectReason, ProfuseGwCapacitySnapshot,
7};
8use saddle_core::{ContextFact, ContextLabel, RequestExecutionView, RequestViewPhase};
9use serde::{Serialize, Serializer, ser::SerializeStruct};
10
11/// Fixed observation facts, not a permit, account or second admission receipt.
12///
13/// ```compile_fail
14/// let facts = saddle_observability::AdmissionCapacityFacts {};
15/// ```
16pub struct AdmissionCapacityFacts {
17    snapshot: ProfuseGwCapacitySnapshot,
18    used: usize,
19    decision: ProfuseGwCapacityDecision,
20    reason: Option<ProfuseGwCapacityRejectReason>,
21    elapsed_ms: u64,
22    resource: Option<saddle_admission::ResourceDecisionSnapshot>,
23}
24impl AdmissionCapacityFacts {
25    pub fn from_receipt(receipt: &ProfuseGwAdmissionObservationReceipt) -> Self {
26        Self {
27            snapshot: receipt.snapshot(),
28            used: receipt.used(),
29            decision: receipt.decision(),
30            reason: receipt.reason(),
31            elapsed_ms: receipt.elapsed_ms(),
32            resource: receipt.resource(),
33        }
34    }
35}
36
37/// Borrow the original view, or report only real pre-root application/lifecycle.
38/// Neither form creates or retains a request root or task.
39///
40/// ```compile_fail
41/// fn retain(view: &saddle_core::RequestExecutionView)
42///     -> saddle_observability::AdmissionEventContext<'static> {
43///     saddle_observability::AdmissionEventContext::Rooted(view)
44/// }
45/// ```
46#[derive(Clone, Copy)]
47pub enum AdmissionEventContext<'a> {
48    Rooted(&'a RequestExecutionView),
49    Unrooted {
50        application: &'a ContextLabel,
51        lifecycle: RequestViewPhase,
52    },
53}
54
55/// Transport facts validated before business admission. The original Core
56/// root has no published call identity yet, so these remain distinct from its
57/// truthful `not_established` context fields.
58#[doc(hidden)]
59#[derive(Clone, Copy, Serialize)]
60pub struct AdmissionRecognizedIngress<'a> {
61    pub request: &'a str,
62    pub route: &'a str,
63    pub process_id: u32,
64    pub process_generation: u128,
65    pub queue_sequence: u64,
66    pub decision_attempt: u64,
67}
68impl Serialize for AdmissionEventContext<'_> {
69    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
70        match self {
71            Self::Rooted(view) => view.serialize(serializer),
72            Self::Unrooted {
73                application,
74                lifecycle,
75            } => {
76                let mut out = serializer.serialize_struct("UnrootedRequestContext", 20)?;
77                out.serialize_field("schema_version", &2u8)?;
78                out.serialize_field("application", &ContextFact::Present(application))?;
79                out.serialize_field("lifecycle", &ContextFact::Present(lifecycle))?;
80                for field in [
81                    "local_request",
82                    "publication",
83                    "call_application",
84                    "module",
85                    "service",
86                    "operation",
87                    "trace_id",
88                    "request",
89                    "span_id",
90                    "route",
91                    "attempt",
92                    "rpc_id",
93                    "zone",
94                    "db_operation",
95                    "scope",
96                    "task",
97                    "target",
98                ] {
99                    out.serialize_field(field, &ContextFact::<()>::NotEstablished)?;
100                }
101                out.end()
102            }
103        }
104    }
105}
106
107/// Construction outcome is independent of the original admission decision.
108/// Concrete error details continue through the original diagnostic mechanism.
109#[derive(Clone, Copy, Debug, Serialize)]
110#[serde(tag = "status", content = "failure", rename_all = "snake_case")]
111pub enum AdmissionConstruction {
112    NotAttempted,
113    QueuedForRetry,
114    Ready,
115    Cancelled,
116    Failed(AdmissionConstructionFailure),
117}
118#[derive(Clone, Copy, Debug, Serialize)]
119#[serde(rename_all = "snake_case")]
120pub enum AdmissionConstructionFailure {
121    Storage,
122    Deadline,
123    RequestIdentity,
124    Context,
125    RejectionTaskBytes,
126    RejectionTaskSlot,
127}
128
129/// Four independent synchronous submissions, in CPU/memory/DB/outbound order.
130/// Enqueued is not written. Partial acceptance is deliberately not collapsed.
131#[must_use]
132pub struct AdmissionCapacitySubmission {
133    pub cpu: DiagnosticSubmission,
134    pub memory: DiagnosticSubmission,
135    pub database: DiagnosticSubmission,
136    pub profuse_contract: DiagnosticSubmission,
137}
138
139#[derive(Serialize)]
140struct CapacityRecord<'a> {
141    schema_version: u8,
142    timestamp_unix_ms: u128,
143    level: &'static str,
144    event: &'static str,
145    stage: &'static str,
146    context: AdmissionEventContext<'a>,
147    recognized_ingress: Option<AdmissionRecognizedIngress<'a>>,
148    capacity_dimension: &'static str,
149    budget: usize,
150    limit: usize,
151    used: usize,
152    bottleneck: &'static str,
153    decision: &'static str,
154    outcome: &'static str,
155    reject_reason: Option<&'static str>,
156    elapsed_ms: u64,
157    resource_source: &'static str,
158    resource: Option<saddle_admission::ResourceDecisionSnapshot>,
159    construction: AdmissionConstruction,
160}
161
162#[derive(Serialize)]
163struct IngressResource<'a> {
164    domain: &'a str,
165    requested_bytes: usize,
166    available_bytes: usize,
167}
168
169#[derive(Serialize)]
170struct IngressResourceDecisionRecord<'a> {
171    schema_version: u8,
172    timestamp_unix_ms: u128,
173    timestamp_role: &'static str,
174    decision_monotonic_ms: Option<u64>,
175    decision_time_valid: bool,
176    decision_time_source: &'static str,
177    resource_valid: bool,
178    level: &'static str,
179    event: &'static str,
180    stage: &'static str,
181    context: AdmissionEventContext<'a>,
182    decision: &'static str,
183    resource_source: &'static str,
184    resource: IngressResource<'a>,
185}
186
187impl EmergencyDiagnosticHandle {
188    /// Emit only byte facts returned by the failed original ledger transaction.
189    /// The error does not carry a monotonic decision instant. The output clock
190    /// below is explicitly not presented as the admission decision clock.
191    /// There is no request root or recognized identity at this accept-time point.
192    pub fn record_ingress_resource_failure(
193        &self,
194        application: &ContextLabel,
195        error: &AdmissionError,
196    ) -> Option<DiagnosticSubmission> {
197        let (domain, requested_bytes, available_bytes) = match error {
198            AdmissionError::FrameworkReserveExceeded { requested, available } =>
199                ("framework", *requested, *available),
200            AdmissionError::ProcessCapacityExceeded { requested, available } =>
201                ("managed", *requested, *available),
202            _ => return None,
203        };
204        let timestamp_unix_ms = std::time::SystemTime::now()
205            .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis();
206        Some(self.submit_fixed_record(&IngressResourceDecisionRecord {
207            schema_version: 1,
208            timestamp_unix_ms,
209            timestamp_role: "output_enqueued",
210            decision_monotonic_ms: None,
211            decision_time_valid: false,
212            decision_time_source: "not_captured_by_ledger_transaction",
213            resource_valid: true,
214            level: "warn",
215            event: "framework.ingress_resource_decision",
216            stage: "admission",
217            context: AdmissionEventContext::Unrooted {
218                application,
219                lifecycle: RequestViewPhase::Admitted,
220            },
221            decision: "capacity_rejected",
222            resource_source: "admission_ledger_same_lock",
223            resource: IngressResource { domain, requested_bytes, available_bytes },
224        }))
225    }
226}
227
228impl Observer {
229    /// Emit original four-dimensional facts without synthetic Call/Event/root.
230    /// Metrics use the existing fixed dimensions even when output is absent.
231    pub fn record_admission_capacity(
232        &self,
233        output: Option<&EmergencyDiagnosticHandle>,
234        context: AdmissionEventContext<'_>,
235        facts: &AdmissionCapacityFacts,
236        construction: AdmissionConstruction,
237    ) -> AdmissionCapacitySubmission {
238        self.record_admission_capacity_with_recognized(output, context, facts, construction, None)
239    }
240
241    #[doc(hidden)]
242    pub fn record_admission_capacity_with_recognized<'a>(
243        &self,
244        output: Option<&EmergencyDiagnosticHandle>,
245        context: AdmissionEventContext<'a>,
246        facts: &AdmissionCapacityFacts,
247        construction: AdmissionConstruction,
248        recognized_ingress: Option<AdmissionRecognizedIngress<'a>>,
249    ) -> AdmissionCapacitySubmission {
250        let snapshot = facts.snapshot;
251        let dimensions = [
252            (CapacityDimension::Cpu, "cpu", snapshot.cpu_budget()),
253            (
254                CapacityDimension::Memory,
255                "memory",
256                snapshot.memory_budget_bytes(),
257            ),
258            (
259                CapacityDimension::Database,
260                "database",
261                snapshot.database_budget(),
262            ),
263            (
264                CapacityDimension::ProfuseContract,
265                "profuse_contract",
266                snapshot.profusecontract_budget(),
267            ),
268        ];
269        let timestamp = std::time::SystemTime::now()
270            .duration_since(std::time::UNIX_EPOCH)
271            .unwrap_or_default()
272            .as_millis();
273        let rejected = facts.decision == ProfuseGwCapacityDecision::CapacityRejected;
274        let submissions = dimensions.map(|(dimension, name, budget)| {
275            self.inner.metrics.capacity(
276                Some(dimension),
277                u64::try_from(facts.used).unwrap_or(u64::MAX),
278                rejected,
279            );
280            let Some(output) = output else {
281                return DiagnosticSubmission::OutputUnavailable;
282            };
283            output.submit_fixed_record(&CapacityRecord {
284                schema_version: 1,
285                timestamp_unix_ms: timestamp,
286                level: "info",
287                event: "framework.capacity",
288                stage: "admission",
289                context,
290                recognized_ingress,
291                capacity_dimension: name,
292                budget,
293                limit: snapshot.active_limit(),
294                used: facts.used,
295                bottleneck: match snapshot.bottleneck() {
296                    ProfuseGwCapacityBottleneck::Cpu => "cpu",
297                    ProfuseGwCapacityBottleneck::Memory => "memory",
298                    ProfuseGwCapacityBottleneck::Database => "database",
299                    ProfuseGwCapacityBottleneck::ProfuseContract => "profuse_contract",
300                },
301                decision: if rejected {
302                    "capacity_rejected"
303                } else {
304                    "accepted"
305                },
306                outcome: if rejected { "rejected" } else { "accepted" },
307                reject_reason: facts.reason.map(|reason| match reason {
308                    ProfuseGwCapacityRejectReason::AtLimit => "at_limit",
309                }),
310                elapsed_ms: facts.elapsed_ms,
311                resource_source: if facts.resource.is_some() {
312                    "admission_ledger_same_lock"
313                } else {
314                    "not_observed_before_resource_check"
315                },
316                resource: facts.resource,
317                construction,
318            })
319        });
320        let [cpu, memory, database, profuse_contract] = submissions;
321        AdmissionCapacitySubmission {
322            cpu,
323            memory,
324            database,
325            profuse_contract,
326        }
327    }
328}
329
330/// Source temporaries only; fixed frame/queued packet are reported by the
331/// existing request_logging_layouts. Neither context form adds an allocation.
332pub fn admission_capacity_layouts() -> [std::alloc::Layout; 4] {
333    [
334        std::alloc::Layout::new::<AdmissionCapacityFacts>(),
335        std::alloc::Layout::new::<AdmissionEventContext<'static>>(),
336        std::alloc::Layout::new::<CapacityRecord<'static>>(),
337        std::alloc::Layout::new::<AdmissionCapacitySubmission>(),
338    ]
339}
saddle_observability/admission.rs

saddle_observability/
admission.rs
