saddle_core/
request_context.rs

1//! Request facts only. Allocation permission and cancellation remain with Runtime/Admission.
2//! Root creation and each local allocation must be preceded by their storage reservation.
3use crate::{CallContext, DbScopeDiagnosticIdentity};
4use serde::{Serialize, Serializer, ser::SerializeStruct};
5use std::sync::{
6    Arc, OnceLock,
7    atomic::{AtomicU64, Ordering},
8};
9
10static NEXT_ROOT: AtomicU64 = AtomicU64::new(1);
11
12/// Closed absence vocabulary, separate from output availability.
13#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
14#[serde(tag = "state", content = "value", rename_all = "snake_case")]
15pub enum ContextFact<T> {
16    Present(T),
17    NotApplicable,
18    NotEstablished,
19    Unavailable,
20}
21
22/// Exact bounded protocol identity. Not an authorization or a user-data container.
23#[derive(Clone, Copy, Eq, PartialEq)]
24pub struct ContextIdentity {
25    bytes: [u8; 256],
26    len: u16,
27}
28impl ContextIdentity {
29    pub fn checked(value: &str) -> Result<Self, ContextConflict> {
30        if value.is_empty() || value.len() > 256 || value.chars().any(char::is_control) {
31            return Err(ContextConflict::InvalidIdentity);
32        }
33        let mut out = Self {
34            bytes: [0; 256],
35            len: value.len() as u16,
36        };
37        out.bytes[..value.len()].copy_from_slice(value.as_bytes());
38        Ok(out)
39    }
40    fn as_str(&self) -> &str {
41        std::str::from_utf8(&self.bytes[..usize::from(self.len)]).expect("validated UTF-8")
42    }
43}
44impl Serialize for ContextIdentity {
45    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
46        s.serialize_str(self.as_str())
47    }
48}
49
50/// Safe metadata, bounded and exact; credentials/URL/query syntax is rejected.
51#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
52#[serde(transparent)]
53pub struct ContextLabel(ContextIdentity);
54impl ContextLabel {
55    pub fn checked(value: &str) -> Result<Self, ContextConflict> {
56        let value = ContextIdentity::checked(value)?;
57        if value.as_str().contains("://")
58            || !value
59                .as_str()
60                .chars()
61                .all(|c| c.is_alphanumeric() || "_.:/{}*-".contains(c))
62        {
63            return Err(ContextConflict::UnsafeMetadata);
64        }
65        Ok(Self(value))
66    }
67}
68
69#[derive(Clone, Copy, Debug, Eq, PartialEq)]
70pub enum ContextConflict {
71    InvalidIdentity,
72    UnsafeMetadata,
73    IdentityGroup,
74    ForeignRoot,
75    ChildRelation,
76    CounterExhausted,
77}
78
79/// Complete validated identity group. Consumed atomically by the entry publisher.
80/// No partial-field setters; rejected publication returns this same group.
81#[derive(Eq, PartialEq)]
82pub struct RequestIdentityGroup {
83    application: ContextLabel,
84    module: ContextLabel,
85    service: ContextLabel,
86    operation: ContextLabel,
87    trace: ContextIdentity,
88    rpc: ContextFact<ContextIdentity>,
89    span: u64,
90    request: ContextIdentity,
91    route: ContextLabel,
92    attempt: u32,
93    zone: ContextFact<ContextLabel>,
94}
95impl RequestIdentityGroup {
96    pub fn from_validated(
97        call: &CallContext,
98        request: &str,
99        route: &str,
100        attempt: u32,
101        zone: ContextFact<ContextLabel>,
102    ) -> Result<Self, ContextConflict> {
103        if attempt == 0 {
104            return Err(ContextConflict::InvalidIdentity);
105        }
106        Ok(Self {
107            application: ContextLabel::checked(call.application().as_str())?,
108            module: ContextLabel::checked(call.module().as_str())?,
109            service: ContextLabel::checked(call.service().as_str())?,
110            operation: ContextLabel::checked(call.operation().as_str())?,
111            trace: ContextIdentity::checked(call.trace_correlation_id().as_str())?,
112            rpc: match call.rpc_correlation_id() {
113                Some(id) => ContextFact::Present(ContextIdentity::checked(id.as_str())?),
114                None => ContextFact::Unavailable,
115            },
116            span: call.span_id().as_u64(),
117            request: ContextIdentity::checked(request)?,
118            route: ContextLabel::checked(route)?,
119            attempt,
120            zone,
121        })
122    }
123}
124
125struct RequestRoot {
126    local: u64,
127    application: ContextLabel,
128    initial: ContextFact<()>,
129    identity: OnceLock<RequestIdentityGroup>,
130}
131impl Drop for RequestRoot {
132    fn drop(&mut self) {
133        observe(self.local, "root_drop");
134    }
135}
136
137/// Unique entry binding capability. Does not own an account or output handle.
138///
139/// ```compile_fail
140/// use saddle_core::RequestRootPublisher;
141/// fn duplicate(p: RequestRootPublisher) { let _ = p.clone(); }
142/// ```
143pub struct RequestRootPublisher {
144    root: Arc<RequestRoot>,
145}
146
147/// Read-only shared reference; cloning shares one allocation, never root fields.
148pub struct RequestRootRef {
149    root: Arc<RequestRoot>,
150}
151impl Clone for RequestRootRef {
152    fn clone(&self) -> Self {
153        observe(self.root.local, "root_share");
154        Self {
155            root: Arc::clone(&self.root),
156        }
157    }
158}
159impl Drop for RequestRootRef {
160    fn drop(&mut self) {
161        observe(self.root.local, "root_release");
162    }
163}
164
165impl RequestRootPublisher {
166    /// Storage-only constructor, not admission. Caller must reserve before calling;
167    /// C supplies layout, R0/A/R supply and enforce the actual reservation seam.
168    pub fn create(
169        application: ContextLabel,
170        initial: ContextFact<()>,
171    ) -> Result<Self, ContextConflict> {
172        if matches!(initial, ContextFact::Present(())) {
173            return Err(ContextConflict::InvalidIdentity);
174        }
175        let local = NEXT_ROOT
176            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1))
177            .map_err(|_| ContextConflict::CounterExhausted)?;
178        let root = Arc::new(RequestRoot {
179            local,
180            application,
181            initial,
182            identity: OnceLock::new(),
183        });
184        observe(local, "root_create");
185        Ok(Self { root })
186    }
187    pub fn reference(&self) -> RequestRootRef {
188        observe(self.root.local, "root_share");
189        RequestRootRef {
190            root: Arc::clone(&self.root),
191        }
192    }
193    #[allow(clippy::result_large_err)] // Recover the fixed input without allocating on rejection.
194    pub fn publish(
195        &mut self,
196        group: RequestIdentityGroup,
197    ) -> Result<(), (ContextConflict, RequestIdentityGroup)> {
198        if group.application != self.root.application {
199            return Err((ContextConflict::IdentityGroup, group));
200        }
201        if let Some(old) = self.root.identity.get() {
202            return if old == &group {
203                Ok(())
204            } else {
205                Err((ContextConflict::IdentityGroup, group))
206            };
207        }
208        self.root
209            .identity
210            .set(group)
211            .map_err(|group| (ContextConflict::IdentityGroup, group))
212    }
213}
214
215/// Closed lifecycle metadata, not a state machine controlling the request.
216#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
217#[serde(rename_all = "snake_case")]
218pub enum RequestViewPhase {
219    SocketAccepted,
220    Reading,
221    Admitted,
222    Dispatch,
223    Handler,
224    Database,
225    Outbound,
226    Response,
227    Finalizing,
228    Finished,
229}
230
231/// Static registered operation labels are borrowed forever, never copied as text.
232#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
233pub struct RegisteredContextOperation(&'static str);
234impl RegisteredContextOperation {
235    pub fn checked(value: &'static str) -> Result<Self, ContextConflict> {
236        ContextLabel::checked(value)?;
237        Ok(Self(value))
238    }
239}
240
241#[derive(Clone, Copy)]
242pub struct RequestLocalFacts {
243    task: ContextFact<u64>,
244    scope: ContextFact<DbScopeDiagnosticIdentity>,
245    db_operation: ContextFact<RegisteredContextOperation>,
246    phase: RequestViewPhase,
247}
248impl RequestLocalFacts {
249    pub fn new(phase: RequestViewPhase) -> Self {
250        Self {
251            task: ContextFact::NotEstablished,
252            scope: ContextFact::NotEstablished,
253            db_operation: ContextFact::NotApplicable,
254            phase,
255        }
256    }
257    /// Numeric observation supplied by the real task owner; not execution authority.
258    pub fn with_task(mut self, task: ContextFact<u64>) -> Self {
259        self.task = task;
260        self
261    }
262    pub fn without_scope(mut self, state: ContextFact<()>) -> Self {
263        self.scope = absent(state);
264        self
265    }
266    pub fn with_db_operation(mut self, operation: RegisteredContextOperation) -> Self {
267        self.db_operation = ContextFact::Present(operation);
268        self
269    }
270}
271
272// Only a real child needs variable local identities. Sibling views share this
273// allocation; the root's original request/trace/zone never appear in this object.
274struct ChildCall {
275    #[cfg(test)]
276    observation: LocalObservation,
277    application: ContextLabel,
278    module: ContextLabel,
279    service: ContextLabel,
280    operation: ContextLabel,
281    rpc: ContextIdentity,
282    span: u64,
283    route: ContextLabel,
284    attempt: u32,
285}
286struct LocalView {
287    #[cfg(test)]
288    observation: LocalObservation,
289    root: Arc<RequestRoot>,
290    published: bool,
291    facts: RequestLocalFacts,
292    child: Option<Arc<ChildCall>>,
293}
294impl Drop for LocalView {
295    fn drop(&mut self) {
296        observe(self.root.local, "view_drop");
297        #[cfg(test)]
298        self.observation.record("destroy");
299    }
300}
301impl Drop for ChildCall {
302    fn drop(&mut self) {
303        #[cfg(test)]
304        self.observation.record("destroy");
305    }
306}
307
308/// Immutable local facts + a frozen visibility stage, sharing the single root.
309/// Source receipts retain this object, not another projection or diagnostic body.
310///
311/// ```compile_fail
312/// use saddle_core::RequestExecutionView;
313/// fn rewrite(view: RequestExecutionView) { view.inner.published = false; }
314/// ```
315pub struct RequestExecutionView {
316    inner: Arc<LocalView>,
317}
318/// A borrowed source-record projection of the actual outbound call target.
319/// It does not change the request identity or retain the target.
320pub struct OutboundSourceContext<'a> {
321    view: &'a RequestExecutionView,
322    business_unit: &'a str,
323}
324impl Clone for RequestExecutionView {
325    fn clone(&self) -> Self {
326        #[cfg(test)]
327        self.inner.observation.record("share");
328        Self {
329            inner: Arc::clone(&self.inner),
330        }
331    }
332}
333impl Drop for RequestExecutionView {
334    fn drop(&mut self) {
335        #[cfg(test)]
336        self.inner.observation.record("release");
337    }
338}
339impl RequestRootRef {
340    pub fn view(&self, facts: RequestLocalFacts) -> RequestExecutionView {
341        RequestExecutionView::allocate(
342            Arc::clone(&self.root),
343            self.root.identity.get().is_some(),
344            facts,
345            None,
346        )
347    }
348    pub fn same_request(&self, view: &RequestExecutionView) -> bool {
349        Arc::ptr_eq(&self.root, &view.inner.root)
350    }
351}
352impl RequestExecutionView {
353    pub fn outbound_source_context<'a>(&'a self, business_unit: &'a str) -> OutboundSourceContext<'a> {
354        OutboundSourceContext { view: self, business_unit }
355    }
356    fn allocate(
357        root: Arc<RequestRoot>,
358        published: bool,
359        facts: RequestLocalFacts,
360        child: Option<Arc<ChildCall>>,
361    ) -> Self {
362        observe(root.local, "view_create");
363        Self {
364            inner: Arc::new(LocalView {
365                #[cfg(test)]
366                observation: LocalObservation::create(root.local, "view"),
367                root,
368                published,
369                facts,
370                child,
371            }),
372        }
373    }
374    /// New operation, no root-text copy or mutation of an earlier view.
375    fn local(&self, facts: RequestLocalFacts) -> Self {
376        Self::allocate(
377            Arc::clone(&self.inner.root),
378            self.inner.published,
379            facts,
380            self.inner.child.clone(),
381        )
382    }
383    pub fn with_phase(&self, phase: RequestViewPhase) -> Self {
384        let mut facts = self.inner.facts;
385        facts.phase = phase;
386        self.local(facts)
387    }
388    pub fn with_db_operation(&self, operation: RegisteredContextOperation) -> Self {
389        self.local(self.inner.facts.with_db_operation(operation))
390    }
391    /// Derive the observation for the actual task selected by the runtime. The
392    /// numeric task is metadata, never a task permit or a sequence minted here.
393    pub fn in_task(&self, task: u64) -> Self {
394        self.local(self.inner.facts.with_task(ContextFact::Present(task)))
395    }
396    pub fn in_db_scope(
397        &self,
398        scope: &crate::DbScopeDiagnosticContext<Self>,
399    ) -> Result<Self, ContextConflict> {
400        let (original, identity) = scope.diagnostic_context();
401        if !self.same_request(original) {
402            return Err(ContextConflict::ForeignRoot);
403        }
404        let mut facts = self.inner.facts;
405        facts.scope = ContextFact::Present(identity);
406        Ok(self.local(facts))
407    }
408    pub fn observed_db_scope(
409        &self,
410        scope: &crate::DbScopeObservation<Self>,
411    ) -> Result<Self, ContextConflict> {
412        let (original, identity) = scope.diagnostic_context();
413        if !self.same_request(original) {
414            return Err(ContextConflict::ForeignRoot);
415        }
416        let mut facts = self.inner.facts;
417        facts.scope = ContextFact::Present(identity);
418        Ok(self.local(facts))
419    }
420    pub fn same_request(&self, other: &Self) -> bool {
421        Arc::ptr_eq(&self.inner.root, &other.inner.root)
422    }
423    pub fn same_view(&self, other: &Self) -> bool {
424        Arc::ptr_eq(&self.inner, &other.inner)
425    }
426    /// Only the existing entry root can authorize a later visibility stage.
427    pub fn refresh(&self, root: &RequestRootRef) -> Result<Self, ContextConflict> {
428        if !root.same_request(self) {
429            return Err(ContextConflict::ForeignRoot);
430        }
431        Ok(Self::allocate(
432            Arc::clone(&self.inner.root),
433            self.inner.root.identity.get().is_some(),
434            self.inner.facts,
435            self.inner.child.clone(),
436        ))
437    }
438    pub fn child(
439        &self,
440        call: &CallContext,
441        request: &str,
442        route: &str,
443        attempt: u32,
444    ) -> Result<Self, ContextConflict> {
445        let identity = self.identity().ok_or(ContextConflict::ChildRelation)?;
446        if identity.trace.as_str() != call.trace_correlation_id().as_str()
447            || identity.request.as_str() != request
448        {
449            return Err(ContextConflict::ForeignRoot);
450        }
451        let parent_rpc = if let Some(child) = &self.inner.child {
452            &child.rpc
453        } else if let ContextFact::Present(rpc) = &identity.rpc {
454            rpc
455        } else {
456            return Err(ContextConflict::ChildRelation);
457        };
458        let rpc = call
459            .rpc_correlation_id()
460            .ok_or(ContextConflict::ChildRelation)?;
461        let suffix = rpc
462            .as_str()
463            .strip_prefix(parent_rpc.as_str())
464            .and_then(|s| s.strip_prefix('.'))
465            .ok_or(ContextConflict::ChildRelation)?;
466        let span = self
467            .inner
468            .child
469            .as_ref()
470            .map_or(identity.span, |child| child.span);
471        if suffix.is_empty()
472            || !suffix.bytes().all(|b| b.is_ascii_digit())
473            || span == call.span_id().as_u64()
474            || attempt == 0
475        {
476            return Err(ContextConflict::ChildRelation);
477        }
478        let child = Arc::new(ChildCall {
479            application: ContextLabel::checked(call.application().as_str())?,
480            module: ContextLabel::checked(call.module().as_str())?,
481            service: ContextLabel::checked(call.service().as_str())?,
482            operation: ContextLabel::checked(call.operation().as_str())?,
483            rpc: ContextIdentity::checked(rpc.as_str())?,
484            span: call.span_id().as_u64(),
485            route: ContextLabel::checked(route)?,
486            attempt,
487            #[cfg(test)]
488            observation: LocalObservation::create(self.inner.root.local, "child"),
489        });
490        Ok(Self::allocate(
491            Arc::clone(&self.inner.root),
492            self.inner.published,
493            self.inner.facts,
494            Some(child),
495        ))
496    }
497    fn identity(&self) -> Option<&RequestIdentityGroup> {
498        self.inner
499            .published
500            .then(|| self.inner.root.identity.get())
501            .flatten()
502    }
503}
504
505fn absent<T>(state: ContextFact<()>) -> ContextFact<T> {
506    match state {
507        ContextFact::NotApplicable => ContextFact::NotApplicable,
508        ContextFact::NotEstablished => ContextFact::NotEstablished,
509        _ => ContextFact::Unavailable,
510    }
511}
512impl Serialize for RequestExecutionView {
513    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
514        self.serialize_context(s, None)
515    }
516}
517impl Serialize for OutboundSourceContext<'_> {
518    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
519        self.view.serialize_context(s, Some(self.business_unit))
520    }
521}
522impl RequestExecutionView {
523    fn serialize_context<S: Serializer>(&self, s: S, business_unit: Option<&str>) -> Result<S::Ok, S::Error> {
524        let mut out = s.serialize_struct("RequestContext", 21 + usize::from(business_unit.is_some()))?;
525        let identity = self.identity();
526        let child = self.inner.child.as_deref();
527        let initial = self.inner.root.initial;
528        out.serialize_field("schema_version", &2u8)?;
529        out.serialize_field("local_request", &self.inner.root.local)?;
530        out.serialize_field("publication", &u8::from(self.inner.published))?;
531        out.serialize_field(
532            "application",
533            &ContextFact::Present(&self.inner.root.application),
534        )?;
535        out.serialize_field(
536            "call_application",
537            &ContextFact::Present(child.map_or(&self.inner.root.application, |c| &c.application)),
538        )?;
539        macro_rules! root_field {
540            ($name:literal, $field:ident) => {
541                out.serialize_field(
542                    $name,
543                    &identity.map_or_else(|| absent(initial), |g| ContextFact::Present(&g.$field)),
544                )?;
545            };
546        }
547        macro_rules! call_field {
548            ($name:literal, $field:ident) => {
549                out.serialize_field(
550                    $name,
551                    &child
552                        .map(|c| &c.$field)
553                        .or_else(|| identity.map(|g| &g.$field))
554                        .map_or_else(|| absent(initial), ContextFact::Present),
555                )?;
556            };
557        }
558        call_field!("module", module);
559        call_field!("service", service);
560        call_field!("operation", operation);
561        root_field!("trace_id", trace);
562        root_field!("request", request);
563        let span = child.map(|c| c.span).or_else(|| identity.map(|g| g.span));
564        out.serialize_field(
565            "span_id",
566            &span.map_or_else(
567                || absent(initial),
568                |s| ContextFact::Present(SpanProjection(s)),
569            ),
570        )?;
571        call_field!("route", route);
572        call_field!("attempt", attempt);
573        let rpc = child
574            .map(|c| ContextFact::Present(c.rpc))
575            .or_else(|| identity.map(|g| g.rpc))
576            .unwrap_or_else(|| absent(initial));
577        out.serialize_field("rpc_id", &rpc)?;
578        out.serialize_field(
579            "zone",
580            &identity.map_or_else(|| absent(initial), |g| g.zone),
581        )?;
582        out.serialize_field("db_operation", &self.inner.facts.db_operation)?;
583        out.serialize_field("scope", &self.inner.facts.scope)?;
584        out.serialize_field("task", &self.inner.facts.task)?;
585        out.serialize_field("lifecycle", &ContextFact::Present(self.inner.facts.phase))?;
586        // Target is the registered route, never endpoint/URL or credentials.
587        out.serialize_field(
588            "target",
589            &child.map_or(ContextFact::NotApplicable, |c| {
590                ContextFact::Present(&c.route)
591            }),
592        )?;
593        if let Some(business_unit) = business_unit {
594            out.serialize_field("business_unit", &ContextFact::Present(business_unit))?;
595        }
596        out.end()
597    }
598}
599
600/// Payload and shared allocation layouts, not a reservation or full task charge.
601/// Arc control block calculation follows std's two-AtomicUsize header, with
602/// Layout::extend padding. Allocator metadata is NOT included; R0 must account it.
603pub fn request_context_layouts() -> [(std::alloc::Layout, std::alloc::Layout); 3] {
604    fn pair<T>() -> (std::alloc::Layout, std::alloc::Layout) {
605        let payload = std::alloc::Layout::new::<T>();
606        let header = std::alloc::Layout::new::<[std::sync::atomic::AtomicUsize; 2]>();
607        (
608            payload,
609            header
610                .extend(payload)
611                .expect("fixed layout")
612                .0
613                .pad_to_align(),
614        )
615    }
616    [
617        pair::<RequestRoot>(),
618        pair::<LocalView>(),
619        pair::<ChildCall>(),
620    ]
621}
622
623#[cfg(not(test))]
624fn observe(_: u64, _: &'static str) {}
625#[cfg(test)]
626fn observe(root: u64, event: &'static str) {
627    EVENTS.lock().unwrap().push((root, event));
628    record_observation(root, root, "root", event);
629}
630#[cfg(test)]
631static EVENTS: std::sync::Mutex<Vec<(u64, &'static str)>> = std::sync::Mutex::new(Vec::new());
632
633struct SpanProjection(u64);
634impl Serialize for SpanProjection {
635    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
636        let mut bytes = [b'0'; 16];
637        for (i, byte) in bytes.iter_mut().enumerate() {
638            *byte = b"0123456789abcdef"[((self.0 >> ((15 - i) * 4)) & 15) as usize];
639        }
640        serializer.serialize_str(std::str::from_utf8(&bytes).expect("hex"))
641    }
642}
643
644// Private observation holds numbers only, never Arc/Weak, callbacks or objects.
645// Instrumentation storage is test-only and excluded from production layouts.
646#[cfg(test)]
647struct LocalObservation {
648    id: u64,
649    root: u64,
650    kind: &'static str,
651}
652#[cfg(test)]
653static LOCAL_EVENTS: std::sync::Mutex<Vec<(u64, u64, &'static str, &'static str)>> =
654    std::sync::Mutex::new(Vec::new());
655#[cfg(test)]
656impl LocalObservation {
657    fn create(root: u64, kind: &'static str) -> Self {
658        static NEXT: AtomicU64 = AtomicU64::new(1);
659        let observation = Self {
660            id: NEXT.fetch_add(1, Ordering::Relaxed),
661            root,
662            kind,
663        };
664        observation.record("create");
665        observation
666    }
667    fn record(&self, event: &'static str) {
668        LOCAL_EVENTS
669            .lock()
670            .unwrap()
671            .push((self.id, self.root, self.kind, event));
672        record_observation(self.root, self.id, self.kind, event);
673    }
674}
675
676#[cfg(test)]
677type ObservationEvent = (u64, u64, u64, &'static str, &'static str);
678#[cfg(test)]
679static ORDERED_EVENTS: std::sync::Mutex<Vec<ObservationEvent>> = std::sync::Mutex::new(Vec::new());
680#[cfg(test)]
681fn record_observation(root: u64, object: u64, kind: &'static str, event: &'static str) {
682    static SEQUENCE: AtomicU64 = AtomicU64::new(1);
683    let sequence = SEQUENCE.fetch_add(1, Ordering::Relaxed);
684    ORDERED_EVENTS
685        .lock()
686        .unwrap()
687        .push((sequence, root, object, kind, event));
688}
689
690#[cfg(test)]
691mod tests;
saddle_core/request_context.rs

saddle_core/
request_context.rs
