query_runtime/observability/
jsonl.rs

1//! JSONL export for sealed runtime observation snapshots.
2
3use std::{error::Error as StdError, fmt, io, io::Write};
4
5use super::{RuntimeObservationRecord, RuntimeTopologyRecord, SealedRuntimeObservationSnapshot};
6
7/// Machine-readable export failure category.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9#[non_exhaustive]
10pub enum RuntimeObservationExportErrorKind {
11    /// A typed record could not be encoded. Retrying unchanged data will not help.
12    InvalidRecord,
13    /// The destination rejected a write.
14    DestinationUnavailable,
15}
16
17/// Retry status derived from the export failure.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum RuntimeObservationExportErrorStatus {
20    /// The record cannot be encoded; unchanged data is not suitable for retry.
21    Permanent,
22    /// The destination rejected a write; a healthy destination may accept the sealed data.
23    Temporary,
24}
25
26#[derive(Debug)]
27enum RuntimeObservationExportErrorSource {
28    Encode(serde_json::Error),
29    Write(io::Error),
30}
31
32impl fmt::Display for RuntimeObservationExportErrorSource {
33    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
34        match self {
35            Self::Encode(error) => error.fmt(formatter),
36            Self::Write(error) => error.fmt(formatter),
37        }
38    }
39}
40
41impl StdError for RuntimeObservationExportErrorSource {
42    fn source(&self) -> Option<&(dyn StdError + 'static)> {
43        match self {
44            Self::Encode(error) => Some(error),
45            Self::Write(error) => Some(error),
46        }
47    }
48}
49
50/// Typed JSONL export error. Export failure never mutates the sealed snapshot.
51#[derive(Debug)]
52pub struct RuntimeObservationExportError {
53    kind: RuntimeObservationExportErrorKind,
54    status: RuntimeObservationExportErrorStatus,
55    record_index: usize,
56    source: RuntimeObservationExportErrorSource,
57}
58
59impl RuntimeObservationExportError {
60    /// Returns the caller-action category.
61    pub const fn kind(&self) -> RuntimeObservationExportErrorKind {
62        self.kind
63    }
64
65    /// Returns whether retrying with a healthy destination can help.
66    pub const fn status(&self) -> RuntimeObservationExportErrorStatus {
67        self.status
68    }
69
70    /// Zero-based JSONL record index that failed.
71    pub const fn record_index(&self) -> usize {
72        self.record_index
73    }
74
75    /// Returns whether the failure is classified as temporary.
76    pub const fn is_retryable(&self) -> bool {
77        matches!(self.status, RuntimeObservationExportErrorStatus::Temporary)
78    }
79
80    fn encode(record_index: usize, source: serde_json::Error) -> Self {
81        if let Some(kind) = source.io_error_kind() {
82            let status = match kind {
83                io::ErrorKind::Interrupted | io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => {
84                    RuntimeObservationExportErrorStatus::Temporary
85                },
86                _ => RuntimeObservationExportErrorStatus::Permanent,
87            };
88            return Self {
89                kind: RuntimeObservationExportErrorKind::DestinationUnavailable,
90                status,
91                record_index,
92                source: RuntimeObservationExportErrorSource::Encode(source),
93            };
94        }
95        Self {
96            kind: RuntimeObservationExportErrorKind::InvalidRecord,
97            status: RuntimeObservationExportErrorStatus::Permanent,
98            record_index,
99            source: RuntimeObservationExportErrorSource::Encode(source),
100        }
101    }
102
103    fn write(record_index: usize, source: io::Error) -> Self {
104        let status = match source.kind() {
105            io::ErrorKind::Interrupted | io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => {
106                RuntimeObservationExportErrorStatus::Temporary
107            },
108            _ => RuntimeObservationExportErrorStatus::Permanent,
109        };
110        Self {
111            kind: RuntimeObservationExportErrorKind::DestinationUnavailable,
112            status,
113            record_index,
114            source: RuntimeObservationExportErrorSource::Write(source),
115        }
116    }
117}
118
119impl fmt::Display for RuntimeObservationExportError {
120    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
121        write!(
122            formatter,
123            "runtime observation JSONL export failed: kind={:?}, status={:?}, record_index={}: {}",
124            self.kind, self.status, self.record_index, self.source
125        )
126    }
127}
128
129impl StdError for RuntimeObservationExportError {
130    fn source(&self) -> Option<&(dyn StdError + 'static)> {
131        Some(&self.source)
132    }
133}
134
135/// Writes a sealed snapshot as ordered, independently parseable JSONL records.
136pub fn write_runtime_observation_jsonl<W: Write>(
137    snapshot: &SealedRuntimeObservationSnapshot,
138    writer: &mut W,
139) -> Result<(), RuntimeObservationExportError> {
140    let data = snapshot.data();
141    let mut record_index = 0;
142    write_record(writer, record_index, &RuntimeObservationRecord::Manifest(data.manifest.clone()))?;
143    record_index += 1;
144
145    for retained in &data.topology {
146        let record = match &retained.record {
147            RuntimeTopologyRecord::PlanNode(row) => {
148                RuntimeObservationRecord::PlanNode { sequence: retained.sequence, row: row.clone() }
149            },
150            RuntimeTopologyRecord::PlanEdge(row) => {
151                RuntimeObservationRecord::PlanEdge { sequence: retained.sequence, row: row.clone() }
152            },
153            RuntimeTopologyRecord::TaskDefinition(row) => {
154                RuntimeObservationRecord::TaskDefinition { sequence: retained.sequence, row: row.clone() }
155            },
156            RuntimeTopologyRecord::CompiledEdge(row) => {
157                RuntimeObservationRecord::CompiledEdge { sequence: retained.sequence, row: row.clone() }
158            },
159            RuntimeTopologyRecord::BarrierDefinition(row) => {
160                RuntimeObservationRecord::BarrierDefinition { sequence: retained.sequence, row: row.clone() }
161            },
162            RuntimeTopologyRecord::SinkDefinition(row) => {
163                RuntimeObservationRecord::SinkDefinition { sequence: retained.sequence, row: row.clone() }
164            },
165            RuntimeTopologyRecord::JobDefinition(row) => {
166                RuntimeObservationRecord::JobDefinition { sequence: retained.sequence, row: row.clone() }
167            },
168            RuntimeTopologyRecord::WorkerDefinition(row) => {
169                RuntimeObservationRecord::WorkerDefinition { sequence: retained.sequence, row: row.clone() }
170            },
171            RuntimeTopologyRecord::Resource(row) => {
172                RuntimeObservationRecord::Resource { sequence: retained.sequence, row: row.clone() }
173            },
174        };
175        write_record(writer, record_index, &record)?;
176        record_index += 1;
177    }
178    for event in &data.events {
179        write_record(writer, record_index, &RuntimeObservationRecord::TimelineEvent(event.clone()))?;
180        record_index += 1;
181    }
182    write_record(writer, record_index, &RuntimeObservationRecord::Footer(data.footer.clone()))
183}
184
185fn write_record<W: Write>(
186    writer: &mut W,
187    record_index: usize,
188    record: &RuntimeObservationRecord,
189) -> Result<(), RuntimeObservationExportError> {
190    // Serialize directly into the destination. Besides avoiding one complete
191    // record-sized allocation, this lets bounded destinations stop serde as
192    // soon as their hard byte limit is reached.
193    serde_json::to_writer(&mut *writer, record)
194        .map_err(|error| RuntimeObservationExportError::encode(record_index, error))?;
195    writer.write_all(b"\n").map_err(|error| RuntimeObservationExportError::write(record_index, error))
196}
query_runtime/observability/jsonl.rs

query_runtime/observability/
jsonl.rs
