hara_native/instrumentation/hub/
delivery.rs

1use std::collections::{BTreeMap, BTreeSet, VecDeque};
2
3use crate::instrumentation::{
4    Capability, EventDelivery, EventEnvelope, EventKind, EventLocation, EventPhase,
5    InstrumentDirective, InstrumentHandle, InstrumentRegistration, ProjectionLimits,
6    ProjectionRequest, TargetDescriptor, TargetHandle, INSTRUMENTATION_EVENT_SCHEMA,
7    INSTRUMENTATION_PROTOCOL,
8};
9
10use super::{ControlLease, InstrumentationError, InstrumentationHub};
11
12#[derive(Debug, Clone, Default, PartialEq, Eq)]
13pub struct PortableProjection {
14    pub kind: String,
15    pub fields: BTreeMap<String, String>,
16}
17
18impl PortableProjection {
19    pub fn new(kind: impl Into<String>) -> Self {
20        Self {
21            kind: kind.into(),
22            fields: BTreeMap::new(),
23        }
24    }
25
26    pub fn with_field(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
27        self.fields.insert(name.into(), value.into());
28        self
29    }
30}
31
32#[derive(Debug, Clone, Default, PartialEq, Eq)]
33pub struct EventProjection {
34    pub current_frame: Option<PortableProjection>,
35    pub frames: Option<PortableProjection>,
36    pub locals: Option<PortableProjection>,
37    pub stack: Option<PortableProjection>,
38    pub value_preview: Option<PortableProjection>,
39    pub machine_snapshot: Option<PortableProjection>,
40}
41
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct ProducerEvent {
44    pub phase: EventPhase,
45    pub event: EventKind,
46    pub data: BTreeMap<String, String>,
47}
48
49impl ProducerEvent {
50    pub fn live(event: EventKind) -> Self {
51        Self {
52            phase: EventPhase::Live,
53            event,
54            data: BTreeMap::new(),
55        }
56    }
57
58    pub fn with_data(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
59        self.data.insert(name.into(), value.into());
60        self
61    }
62}
63
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct DeliveredEvent {
66    pub envelope: EventEnvelope,
67    pub projection: EventProjection,
68    /// Cumulative events discarded from this instrument queue before this
69    /// event was retained. Callback deliveries always report zero.
70    pub dropped_before: u64,
71}
72
73#[derive(Debug, Clone, Default, PartialEq, Eq)]
74pub struct EventBatch {
75    pub events: Vec<DeliveredEvent>,
76    pub dropped: u64,
77}
78
79#[derive(Debug, Clone, Default, PartialEq, Eq)]
80pub struct DispatchReport {
81    /// Callback delivery is returned to the trusted native caller. The hub
82    /// never executes arbitrary guest code on the application thread.
83    pub callbacks: Vec<DeliveredEvent>,
84    pub queued: usize,
85    pub dropped: usize,
86}
87
88/// Borrowed, lazy inspection over one authoritative execution object.
89/// Producers may return `None` for a projection not supported by their target.
90pub trait EventAccess {
91    fn source_location(&mut self) -> Option<EventLocation> {
92        None
93    }
94
95    fn current_frame(&mut self, _limits: ProjectionLimits) -> Option<PortableProjection> {
96        None
97    }
98
99    fn frames(&mut self, _limits: ProjectionLimits) -> Option<PortableProjection> {
100        None
101    }
102
103    fn locals(&mut self, _limits: ProjectionLimits) -> Option<PortableProjection> {
104        None
105    }
106
107    fn stack(&mut self, _limits: ProjectionLimits) -> Option<PortableProjection> {
108        None
109    }
110
111    fn value_preview(&mut self, _limits: ProjectionLimits) -> Option<PortableProjection> {
112        None
113    }
114
115    fn machine_snapshot(&mut self, _limits: ProjectionLimits) -> Option<PortableProjection> {
116        None
117    }
118}
119
120#[derive(Debug, Default)]
121pub(super) struct DeliveryState {
122    queues: BTreeMap<InstrumentHandle, InstrumentQueue>,
123    sequences: BTreeMap<TargetHandle, u64>,
124    directives: BTreeMap<TargetHandle, InstrumentDirective>,
125}
126
127#[derive(Debug)]
128struct InstrumentQueue {
129    capacity: usize,
130    events: VecDeque<DeliveredEvent>,
131    dropped: u64,
132}
133
134impl InstrumentQueue {
135    fn new(capacity: usize) -> Self {
136        Self {
137            capacity,
138            events: VecDeque::with_capacity(capacity),
139            dropped: 0,
140        }
141    }
142
143    fn push(&mut self, mut event: DeliveredEvent) -> bool {
144        let dropped = if self.events.len() == self.capacity {
145            self.events.pop_front();
146            self.dropped = self.dropped.saturating_add(1);
147            true
148        } else {
149            false
150        };
151        event.dropped_before = self.dropped;
152        self.events.push_back(event);
153        dropped
154    }
155
156    fn drain(&mut self) -> EventBatch {
157        let events = self.events.drain(..).collect();
158        let dropped = std::mem::take(&mut self.dropped);
159        EventBatch { events, dropped }
160    }
161}
162
163impl DeliveryState {
164    pub(super) fn remove_instrument(&mut self, instrument: &InstrumentHandle) {
165        self.queues.remove(instrument);
166    }
167
168    pub(super) fn remove_directive(&mut self, target: &TargetHandle) {
169        self.directives.remove(target);
170    }
171
172    pub(super) fn remove_target(&mut self, target: &TargetHandle) {
173        self.sequences.remove(target);
174        self.directives.remove(target);
175        for queue in self.queues.values_mut() {
176            queue
177                .events
178                .retain(|event| event.envelope.target_id != target.target_id());
179        }
180    }
181
182    pub(super) fn clear(&mut self) {
183        self.queues.clear();
184        self.sequences.clear();
185        self.directives.clear();
186    }
187}
188
189impl InstrumentDirective {
190    pub const fn required_capability(self) -> Option<Capability> {
191        match self {
192            Self::Continue => None,
193            Self::Suspend => Some(Capability::ControlPause),
194            Self::StepNext => Some(Capability::ControlSingleStep),
195            Self::Terminate => Some(Capability::ControlTerminate),
196        }
197    }
198}
199
200impl ProjectionRequest {
201    pub fn is_empty(&self) -> bool {
202        !self.source_location
203            && self.current_frame.is_none()
204            && self.frames.is_none()
205            && self.locals.is_none()
206            && self.stack.is_none()
207            && self.value_preview.is_none()
208            && self.machine_snapshot.is_none()
209    }
210
211    pub fn needs_interpreter_environment(&self) -> bool {
212        self.current_frame.is_some() || self.frames.is_some()
213    }
214
215    pub fn merge_from(&mut self, other: &Self) {
216        self.source_location |= other.source_location;
217        self.current_frame = merge_limits(self.current_frame, other.current_frame);
218        self.frames = merge_limits(self.frames, other.frames);
219        self.locals = merge_limits(self.locals, other.locals);
220        self.stack = merge_limits(self.stack, other.stack);
221        self.value_preview = merge_limits(self.value_preview, other.value_preview);
222        self.machine_snapshot = merge_limits(self.machine_snapshot, other.machine_snapshot);
223    }
224}
225
226impl InstrumentationHub {
227    pub fn target_descriptor(
228        &self,
229        target: &TargetHandle,
230    ) -> Result<&TargetDescriptor, InstrumentationError> {
231        Ok(&self.resolve_target(target)?.descriptor)
232    }
233
234    pub fn instrument_registration(
235        &self,
236        instrument: &InstrumentHandle,
237    ) -> Result<&InstrumentRegistration, InstrumentationError> {
238        Ok(&self.resolve_instrument(instrument)?.registration)
239    }
240
241    /// Returns the aggregate projection required by matching subscriptions for
242    /// one target/event. Producers use this before executing a safepoint so
243    /// environment or frame capture is enabled only when it can be consumed.
244    pub fn requested_projection(
245        &self,
246        target: &TargetHandle,
247        event: EventKind,
248    ) -> Result<ProjectionRequest, InstrumentationError> {
249        self.resolve_target(target)?;
250        if !self.enabled_events.contains(event) {
251            return Ok(ProjectionRequest::default());
252        }
253        let mut projection = ProjectionRequest::default();
254        for attachment in self.attachments_for_target(target)? {
255            let record = self.resolve_instrument(&attachment.instrument)?;
256            if record.registration.events.contains(&event) {
257                projection.merge_from(&record.registration.projection);
258            }
259        }
260        Ok(projection)
261    }
262
263    /// Delivers one cheap producer event to every matching attachment in
264    /// deterministic registration order. Projection methods are never called
265    /// until an attachment has matched both target and event.
266    pub fn emit<A: EventAccess>(
267        &mut self,
268        target: &TargetHandle,
269        event: ProducerEvent,
270        access: &mut A,
271    ) -> Result<DispatchReport, InstrumentationError> {
272        if !self.enabled_events.contains(event.event) {
273            self.resolve_target(target)?;
274            return Ok(DispatchReport::default());
275        }
276
277        let descriptor = self.resolve_target(target)?.descriptor.clone();
278        let subscriptions = self
279            .attachments_for_target(target)?
280            .into_iter()
281            .filter_map(|attachment| {
282                let record = self.resolve_instrument(&attachment.instrument).ok()?;
283                record.registration.events.contains(&event.event).then(|| {
284                    (
285                        attachment.instrument.clone(),
286                        record.registration.projection.clone(),
287                        record.registration.delivery.clone(),
288                    )
289                })
290            })
291            .collect::<Vec<_>>();
292        if subscriptions.is_empty() {
293            return Ok(DispatchReport::default());
294        }
295
296        let sequence = self.delivery.sequences.entry(target.clone()).or_insert(0);
297        *sequence = sequence.saturating_add(1);
298        let sequence = *sequence;
299        let mut report = DispatchReport::default();
300
301        for (instrument, request, delivery) in subscriptions {
302            let (location, projection) = materialize(&request, access);
303            let delivered = DeliveredEvent {
304                envelope: EventEnvelope {
305                    schema: INSTRUMENTATION_EVENT_SCHEMA.into(),
306                    protocol: INSTRUMENTATION_PROTOCOL.into(),
307                    instrument_id: instrument.instrument_id().into(),
308                    runtime: descriptor.backend.clone(),
309                    session_id: descriptor.session_id.clone(),
310                    target_id: descriptor.target_id.clone(),
311                    target_kind: descriptor.kind,
312                    generation: target.generation(),
313                    sequence,
314                    phase: event.phase,
315                    event: event.event,
316                    location,
317                    data: event.data.clone(),
318                },
319                projection,
320                dropped_before: 0,
321            };
322            match delivery {
323                EventDelivery::Callback => report.callbacks.push(delivered),
324                EventDelivery::Queue { capacity } => {
325                    let queue = self
326                        .delivery
327                        .queues
328                        .entry(instrument)
329                        .or_insert_with(|| InstrumentQueue::new(capacity));
330                    debug_assert_eq!(queue.capacity, capacity);
331                    if queue.push(delivered) {
332                        report.dropped = report.dropped.saturating_add(1);
333                    }
334                    report.queued = report.queued.saturating_add(1);
335                }
336            }
337        }
338        Ok(report)
339    }
340
341    pub fn drain_events(
342        &mut self,
343        instrument: &InstrumentHandle,
344    ) -> Result<EventBatch, InstrumentationError> {
345        let registration = self.resolve_instrument(instrument)?.registration.clone();
346        let EventDelivery::Queue { capacity } = registration.delivery else {
347            return Ok(EventBatch::default());
348        };
349        Ok(self
350            .delivery
351            .queues
352            .entry(instrument.clone())
353            .or_insert_with(|| InstrumentQueue::new(capacity))
354            .drain())
355    }
356
357    pub fn queued_event_count(
358        &self,
359        instrument: &InstrumentHandle,
360    ) -> Result<usize, InstrumentationError> {
361        self.resolve_instrument(instrument)?;
362        Ok(self
363            .delivery
364            .queues
365            .get(instrument)
366            .map_or(0, |queue| queue.events.len()))
367    }
368
369    pub fn authorize_control(
370        &self,
371        lease: &ControlLease,
372        capability: Capability,
373    ) -> Result<(), InstrumentationError> {
374        let instrument = self.resolve_instrument(&lease.instrument)?;
375        let target = self.resolve_target(&lease.target)?;
376        match self.control_leases.get(&lease.target) {
377            Some(holder) if holder == &lease.instrument => {}
378            _ => {
379                return Err(InstrumentationError::InvalidControlLease {
380                    target_id: lease.target.target_id().into(),
381                    instrument_id: lease.instrument.instrument_id().into(),
382                });
383            }
384        }
385        if !capability.is_control() || !instrument.registration.capabilities.contains(&capability) {
386            return Err(InstrumentationError::UnsupportedCapabilities {
387                target_id: target.descriptor.target_id.clone(),
388                backend: target.descriptor.backend.clone(),
389                missing: BTreeSet::from([capability]),
390            });
391        }
392        Ok(())
393    }
394
395    /// Stores a controller directive for application at the next real
396    /// evaluator or machine safepoint. `Continue` clears a pending directive.
397    pub fn request_directive(
398        &mut self,
399        lease: &ControlLease,
400        directive: InstrumentDirective,
401    ) -> Result<(), InstrumentationError> {
402        if let Some(capability) = directive.required_capability() {
403            self.authorize_control(lease, capability)?;
404            self.delivery
405                .directives
406                .insert(lease.target.clone(), directive);
407        } else {
408            self.authorize_lease(lease)?;
409            self.delivery.directives.remove(&lease.target);
410        }
411        Ok(())
412    }
413
414    pub fn take_directive(
415        &mut self,
416        target: &TargetHandle,
417    ) -> Result<InstrumentDirective, InstrumentationError> {
418        self.resolve_target(target)?;
419        Ok(self
420            .delivery
421            .directives
422            .remove(target)
423            .unwrap_or(InstrumentDirective::Continue))
424    }
425
426    fn authorize_lease(&self, lease: &ControlLease) -> Result<(), InstrumentationError> {
427        self.resolve_instrument(&lease.instrument)?;
428        self.resolve_target(&lease.target)?;
429        match self.control_leases.get(&lease.target) {
430            Some(holder) if holder == &lease.instrument => Ok(()),
431            _ => Err(InstrumentationError::InvalidControlLease {
432                target_id: lease.target.target_id().into(),
433                instrument_id: lease.instrument.instrument_id().into(),
434            }),
435        }
436    }
437}
438
439fn merge_limits(
440    left: Option<ProjectionLimits>,
441    right: Option<ProjectionLimits>,
442) -> Option<ProjectionLimits> {
443    match (left, right) {
444        (None, value) | (value, None) => value,
445        (Some(left), Some(right)) => Some(ProjectionLimits {
446            max_items: left.max_items.max(right.max_items),
447            max_depth: left.max_depth.max(right.max_depth),
448            max_bytes: left.max_bytes.max(right.max_bytes),
449        }),
450    }
451}
452
453fn materialize<A: EventAccess>(
454    request: &ProjectionRequest,
455    access: &mut A,
456) -> (Option<EventLocation>, EventProjection) {
457    let location = request
458        .source_location
459        .then(|| access.source_location())
460        .flatten();
461    let projection = EventProjection {
462        current_frame: request
463            .current_frame
464            .and_then(|limits| access.current_frame(limits)),
465        frames: request.frames.and_then(|limits| access.frames(limits)),
466        locals: request.locals.and_then(|limits| access.locals(limits)),
467        stack: request.stack.and_then(|limits| access.stack(limits)),
468        value_preview: request
469            .value_preview
470            .and_then(|limits| access.value_preview(limits)),
471        machine_snapshot: request
472            .machine_snapshot
473            .and_then(|limits| access.machine_snapshot(limits)),
474    };
475    (location, projection)
476}
hara_native/instrumentation/hub/delivery.rs

hara_native/instrumentation/hub/
delivery.rs
