query_runtime/observability/recorder/
mod.rs

1//! Query-scoped bounded synchronous runtime observation recorder.
2
3use std::{collections::BTreeMap, fmt, sync::Arc};
4
5use query_planning::correlation::{QueryId, TraceId};
6
7use super::{
8    CleanupFailureSummary, ObservationSequence, PlanExecutionId, QueryAttemptReport, QueryBoundaryKind,
9    RuntimeObservationClassLoss, RuntimeObservationClock, RuntimeObservationFooter, RuntimeObservationLimits,
10    RuntimeObservationLossReason, RuntimeObservationManifest, RuntimeObservationPayload, RuntimeObservationReasonLoss,
11    RuntimeObservationRecordClass, RuntimeSourceLocationId, RuntimeSpanId, RuntimeTerminalOutcome,
12    RuntimeTimelineEvent, RuntimeTopologyRecord, SystemRuntimeObservationClock,
13    metrics_summary::{
14        MetricsCaptureMode, MetricsCompleteness, QueryMetricsCheckpoint, QueryMetricsSummary, SourceReadFact,
15    },
16};
17use crate::{
18    metrics::{QueryMetrics, SourceReadRuntimeMetrics},
19    observability::{ExecutionProgress, RUNTIME_OBSERVATION_JSONL_SCHEMA, RuntimeClockDomain},
20};
21
22mod digest;
23mod handle;
24mod owner;
25mod session;
26mod state;
27
28#[cfg(any(test, feature = "test-hooks"))]
29use digest::TestSourceLocationDigestKeyProvider;
30use digest::{
31    OsSourceLocationDigestKeyProvider, SourceLocationDigestKeyProvider, SourceLocationDigestKeyState,
32    SourceLocationDigestKeyUnavailable,
33};
34#[cfg(test)]
35pub(crate) use owner::RuntimeObservationDeliveryGate;
36use owner::RuntimeObservationDeliveryGuard;
37pub use owner::RuntimeObservationOwner;
38pub(crate) use owner::in_runtime_observation_delivery;
39#[cfg(any(test, feature = "test-hooks"))]
40pub(crate) use session::AttemptReportProvider;
41pub use session::RuntimeObservationSession;
42pub(crate) use session::{AttemptReportSealProvider, QueryCleanupSealDelivery};
43use session::{DeferredRuntimeObservationSeal, RuntimeObservationSealObserver, RuntimeObservationSessionInner};
44pub use state::BudgetWindowPeaks;
45pub(crate) use state::{BudgetWindow, CleanupFailureTally};
46use state::{PlanExecutionMetrics, RuntimeObservationRecorderState};
47
48/// Query-local recorder configuration.
49///
50/// [`Default`] disables capture. [`Self::native_execution`] enables the bounded
51/// native profile. Record and byte caps limit retained detail independently of
52/// the terminal outcome and the exact source totals.
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub struct RuntimeObservationConfig {
55    enabled: bool,
56    limits: RuntimeObservationLimits,
57}
58
59impl RuntimeObservationConfig {
60    /// Default bounded capture used by native execution adapters.
61    pub const NATIVE_EXECUTION_MAX_RECORDS: u64 = 65_536;
62    /// Deterministic logical-byte cap paired with [`Self::NATIVE_EXECUTION_MAX_RECORDS`].
63    pub const NATIVE_EXECUTION_MAX_BYTES: u64 = 8 * 1024 * 1024;
64
65    /// Disables runtime recording.
66    pub const fn disabled() -> Self {
67        Self { enabled: false, limits: RuntimeObservationLimits { max_retained_records: 0, max_retained_bytes: 0 } }
68    }
69
70    /// Enables recording with deterministic record and logical-byte caps.
71    /// Zero caps permit recording attempts but retain no records.
72    pub const fn bounded(max_retained_records: u64, max_retained_bytes: u64) -> Self {
73        Self { enabled: true, limits: RuntimeObservationLimits { max_retained_records, max_retained_bytes } }
74    }
75
76    /// Enables the bounded per-query native execution capture profile.
77    pub const fn native_execution() -> Self {
78        Self::bounded(Self::NATIVE_EXECUTION_MAX_RECORDS, Self::NATIVE_EXECUTION_MAX_BYTES)
79    }
80
81    /// Returns whether payload construction and recording are enabled.
82    pub(crate) const fn is_enabled(self) -> bool {
83        self.enabled
84    }
85
86    /// Returns the configured retention limits.
87    pub const fn limits(self) -> RuntimeObservationLimits {
88        self.limits
89    }
90}
91
92impl Default for RuntimeObservationConfig {
93    fn default() -> Self {
94        Self::disabled()
95    }
96}
97
98/// Optional event identity attached outside the payload.
99#[derive(Debug, Clone, Default, PartialEq, Eq)]
100pub struct RuntimeEventContext {
101    pub(crate) plan_execution_id: Option<PlanExecutionId>,
102    pub(crate) span_id: Option<RuntimeSpanId>,
103    pub(crate) parent_span_id: Option<RuntimeSpanId>,
104}
105
106impl RuntimeEventContext {
107    /// Creates context for one plan execution.
108    pub fn for_plan(plan_execution_id: PlanExecutionId) -> Self {
109        Self { plan_execution_id: Some(plan_execution_id), ..Self::default() }
110    }
111
112    /// Attaches one span and its optional parent.
113    pub const fn with_span(mut self, span_id: RuntimeSpanId, parent_span_id: Option<RuntimeSpanId>) -> Self {
114        self.span_id = Some(span_id);
115        self.parent_span_id = parent_span_id;
116        self
117    }
118}
119
120/// Infallible result of one hot-path record attempt.
121#[derive(Debug, Clone, Copy, PartialEq, Eq)]
122pub enum RuntimeObservationRecordDisposition {
123    Disabled,
124    Retained { sequence: ObservationSequence, logical_bytes: u64 },
125    Dropped { sequence: ObservationSequence, reason: RuntimeObservationLossReason },
126    Sealed,
127}
128
129/// Immutable live snapshot. Its footer is always incomplete and has no terminal outcome.
130#[derive(Debug, Clone)]
131pub struct LiveRuntimeObservationSnapshot {
132    data: Arc<RuntimeObservationSnapshotData>,
133}
134
135/// Immutable sealed snapshot accepted by the JSONL exporter.
136#[derive(Debug, Clone)]
137pub struct SealedRuntimeObservationSnapshot {
138    data: Arc<RuntimeObservationSnapshotData>,
139}
140
141#[derive(Debug, Clone)]
142pub(crate) struct RuntimeObservationSnapshotData {
143    pub(crate) manifest: RuntimeObservationManifest,
144    pub(crate) topology: Vec<RetainedTopologyRecord>,
145    pub(crate) events: Vec<RuntimeTimelineEvent>,
146    pub(crate) footer: RuntimeObservationFooter,
147    /// Each plan execution's frozen metrics. Only a seal fills it.
148    pub(crate) plans: BTreeMap<PlanExecutionId, QueryMetrics>,
149}
150
151#[derive(Debug, Clone, PartialEq, Eq)]
152pub(crate) struct RetainedTopologyRecord {
153    pub(crate) sequence: ObservationSequence,
154    pub(crate) record: RuntimeTopologyRecord,
155}
156
157macro_rules! snapshot_accessors {
158    ($snapshot:ident) => {
159        impl $snapshot {
160            /// Immutable session manifest.
161            pub fn manifest(&self) -> &RuntimeObservationManifest {
162                &self.data.manifest
163            }
164
165            /// Static topology sorted by typed row identity.
166            pub fn topology(&self) -> impl ExactSizeIterator<Item = &RuntimeTopologyRecord> {
167                self.data.topology.iter().map(|record| &record.record)
168            }
169
170            /// Dynamic events sorted by recorder sequence.
171            pub fn events(&self) -> &[RuntimeTimelineEvent] {
172                &self.data.events
173            }
174
175            /// Completeness, terminal, loss, and degradation metadata.
176            pub fn footer(&self) -> &RuntimeObservationFooter {
177                &self.data.footer
178            }
179
180            /// Fixed execution-attempt summary, when this observation belongs to an attempt.
181            ///
182            /// The embedded report is the same immutable value delivered by
183            /// the attempt completion. In particular, `report_ready_ns` is
184            /// fixed before this snapshot is published and the JSONL footer
185            /// serializes this exact allocation.
186            pub fn attempt_report(&self) -> Option<&QueryAttemptReport> {
187                self.data.footer.attempt_report.as_deref()
188            }
189        }
190    };
191}
192
193snapshot_accessors!(LiveRuntimeObservationSnapshot);
194snapshot_accessors!(SealedRuntimeObservationSnapshot);
195
196impl SealedRuntimeObservationSnapshot {
197    pub(crate) fn data(&self) -> &RuntimeObservationSnapshotData {
198        &self.data
199    }
200
201    /// The metrics of plan execution `plan`, as frozen when its owner
202    /// projected them or, failing that, when this session sealed.
203    /// `None` means that the snapshot has no frozen metrics for that invocation.
204    pub fn query_metrics(&self, plan: &PlanExecutionId) -> Option<&QueryMetrics> {
205        self.data.plans.get(plan)
206    }
207
208    /// Returns whether both values are views of the same immutable seal.
209    ///
210    /// Query identifiers are intentionally insufficient for this comparison.
211    /// Independently created recorder sessions may share one external query
212    /// identity. The comparison uses the `Arc` allocation behind each snapshot.
213    /// Only test code calls this method.
214    #[cfg(test)]
215    pub(crate) fn is_same_snapshot(&self, other: &Self) -> bool {
216        Arc::ptr_eq(&self.data, &other.data)
217    }
218}
219
220/// Cheaply cloneable recording capability without terminal-seal authority.
221#[derive(Clone, Default)]
222pub struct RuntimeObservationHandle {
223    config: RuntimeObservationConfig,
224    inner: Option<Arc<RuntimeObservationSessionInner>>,
225}
226
227impl fmt::Debug for RuntimeObservationHandle {
228    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
229        formatter
230            .debug_struct("RuntimeObservationHandle")
231            .field("enabled", &self.is_enabled())
232            .field("limits", &self.config.limits())
233            .finish_non_exhaustive()
234    }
235}
236
237#[cfg(test)]
238mod tests;
query_runtime/observability/recorder/mod.rs

query_runtime/observability/recorder/
mod.rs
