solti_model/domain/output.rs
1//! # Task output
2//!
3//! [`OutputEvent`] is the shared live-output event.
4//! [`OutputChunk`] carries one binary stdout or stderr chunk.
5//!
6//! This module defines data and serde encoding only.
7//! Publishers, channels, retention, and subscriptions belong to higher layers.
8//!
9//! ## Serde Contract
10//!
11//! Events use a flat `type` tag.
12//! Timestamps use Unix milliseconds.
13//! Chunk bytes use standard padded base64.
14//!
15//! ```text
16//! OutputEvent
17//! ├── chunk ──▶ generation, attempt, stream, seq, ts, line
18//! ├── runStarted ──▶ generation, attempt, startedAt
19//! ├── runFinished ──▶ generation, attempt, exitCode, finishedAt
20//! └── lagged ──▶ skipped
21//! ```
22//!
23//! `solti-api` maps the same domain events to the separate protobuf shape.
24
25use std::time::SystemTime;
26
27use bytes::Bytes;
28use serde::{Deserialize, Serialize};
29
30/// Standard stream that produced a chunk.
31///
32/// ## Example
33///
34/// ```
35/// use solti_model::StreamKind;
36///
37/// let json = serde_json::to_string(&StreamKind::Stdout).unwrap();
38/// assert_eq!(json, r#""stdout""#);
39/// ```
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
41#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
42#[serde(rename_all = "lowercase")]
43pub enum StreamKind {
44 /// Standard output (`stdout`).
45 Stdout,
46 /// Standard error (`stderr`).
47 Stderr,
48}
49
50/// Event in a task live-output stream.
51///
52/// Run markers are the best effort.
53/// They are not ordering barriers for chunks.
54///
55/// ## JSON Shape
56///
57/// ```text
58/// {"type":"chunk","generation":2,"attempt":1,"stream":"stdout","seq":0,"ts":1700,"line":"aGVsbG8="}
59/// {"type":"runStarted","generation":2,"attempt":1,"startedAt":1700}
60/// {"type":"runFinished","generation":2,"attempt":1,"exitCode":0,"finishedAt":1701}
61/// {"type":"lagged","skipped":42}
62/// ```
63///
64/// ## Example
65///
66/// ```
67/// use bytes::Bytes;
68/// use solti_model::{OutputChunk, OutputEvent, StreamKind};
69/// use std::time::SystemTime;
70///
71/// let event = OutputEvent::Chunk(OutputChunk {
72/// generation: 2,
73/// attempt: 1,
74/// stream: StreamKind::Stdout,
75/// seq: 0,
76/// ts: SystemTime::UNIX_EPOCH,
77/// line: Bytes::from_static(b"hello"),
78/// });
79///
80/// let json = serde_json::to_string(&event).unwrap();
81/// assert!(json.contains(r#""type":"chunk""#));
82/// ```
83#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
84#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
85#[serde(tag = "type", rename_all = "camelCase")]
86#[non_exhaustive]
87pub enum OutputEvent {
88 /// Carries stdout or stderr bytes from one run.
89 Chunk(OutputChunk),
90
91 /// Reports that a run attempt started.
92 ///
93 /// This marker is not an ordering barrier for chunks.
94 /// Use each chunk's `generation`, `attempt`, `stream`, and `seq` fields for grouping and ordering.
95 #[serde(rename_all = "camelCase")]
96 RunStarted {
97 /// Desired-state generation executed by this run.
98 generation: u64,
99 /// Attempt number of the run that just started.
100 attempt: u32,
101 /// Wall-clock start time (unix milliseconds on the wire).
102 #[serde(with = "crate::resource::metadata::time_serde")]
103 #[cfg_attr(
104 feature = "schema",
105 schemars(schema_with = "crate::schema::unix_millis")
106 )]
107 started_at: SystemTime,
108 },
109
110 /// Reports that a run attempt finished.
111 ///
112 /// This marker is not an ordering barrier: chunks for the same generation and attempt may still be observed after it.
113 #[serde(rename_all = "camelCase")]
114 RunFinished {
115 /// Desired-state generation executed by this run.
116 generation: u64,
117 /// Attempt number of the run that finished.
118 attempt: u32,
119 /// Process exit code.
120 ///
121 /// `None` means no exit code was available.
122 #[serde(skip_serializing_if = "Option::is_none")]
123 exit_code: Option<i32>,
124 /// Wall-clock finish time (unix milliseconds on the wire).
125 #[serde(with = "crate::resource::metadata::time_serde")]
126 #[cfg_attr(
127 feature = "schema",
128 schemars(schema_with = "crate::schema::unix_millis")
129 )]
130 finished_at: SystemTime,
131 },
132
133 /// Reports events lost before the next delivered event.
134 Lagged {
135 /// Number of lost events.
136 skipped: u64,
137 },
138}
139
140/// Output bytes from one task run.
141///
142/// ## Example
143///
144/// ```
145/// use bytes::Bytes;
146/// use solti_model::{OutputChunk, StreamKind};
147/// use std::time::SystemTime;
148///
149/// let chunk = OutputChunk {
150/// generation: 2,
151/// attempt: 1,
152/// stream: StreamKind::Stderr,
153/// seq: 7,
154/// ts: SystemTime::UNIX_EPOCH,
155/// line: Bytes::from_static(b"warning"),
156/// };
157///
158/// assert_eq!(chunk.seq, 7);
159/// ```
160#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
161#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
162#[serde(rename_all = "camelCase")]
163pub struct OutputChunk {
164 /// Desired-state generation this chunk belongs to.
165 pub generation: u64,
166 /// Attempt that produced this chunk.
167 ///
168 /// [`TaskRun::attempt`]: crate::TaskRun::attempt
169 pub attempt: u32,
170 /// Standard stream that produced the chunk.
171 pub stream: StreamKind,
172 /// Sequence number within this generation, attempt, and stream.
173 ///
174 /// Producers define allocation and reset behavior.
175 pub seq: u64,
176 /// Wall-clock event time.
177 ///
178 /// Serde encodes it as Unix milliseconds.
179 #[serde(with = "crate::resource::metadata::time_serde")]
180 #[cfg_attr(
181 feature = "schema",
182 schemars(schema_with = "crate::schema::unix_millis")
183 )]
184 pub ts: SystemTime,
185 /// Raw output bytes.
186 ///
187 /// Serde encodes them as standard padded base64.
188 #[serde(with = "bytes_as_base64")]
189 #[cfg_attr(
190 feature = "schema",
191 schemars(schema_with = "crate::schema::base64_bytes")
192 )]
193 pub line: Bytes,
194}
195
196/// Serde adapter for exact binary round trips through JSON.
197mod bytes_as_base64 {
198 use base64::{Engine as _, engine::general_purpose::STANDARD};
199 use bytes::Bytes;
200 use serde::{Deserialize, Deserializer, Serializer};
201
202 pub(super) fn serialize<S>(b: &Bytes, s: S) -> Result<S::Ok, S::Error>
203 where
204 S: Serializer,
205 {
206 s.serialize_str(&STANDARD.encode(b))
207 }
208
209 pub(super) fn deserialize<'de, D>(d: D) -> Result<Bytes, D::Error>
210 where
211 D: Deserializer<'de>,
212 {
213 let s = String::deserialize(d)?;
214 STANDARD
215 .decode(s)
216 .map(Bytes::from)
217 .map_err(serde::de::Error::custom)
218 }
219}
220
221#[cfg(test)]
222mod tests {
223 use super::*;
224
225 use std::time::{Duration, UNIX_EPOCH};
226
227 #[test]
228 fn wire_shape_is_pinned_for_every_event() {
229 let cases = [
230 (
231 OutputEvent::Chunk(OutputChunk {
232 generation: 2,
233 attempt: 1,
234 stream: StreamKind::Stdout,
235 seq: 0,
236 ts: UNIX_EPOCH + Duration::from_millis(1_700),
237 line: Bytes::from_static(b"hi"),
238 }),
239 r#"{"type":"chunk","generation":2,"attempt":1,"stream":"stdout","seq":0,"ts":1700,"line":"aGk="}"#,
240 ),
241 (
242 OutputEvent::RunStarted {
243 generation: 4,
244 attempt: 2,
245 started_at: UNIX_EPOCH + Duration::from_millis(1_234),
246 },
247 r#"{"type":"runStarted","generation":4,"attempt":2,"startedAt":1234}"#,
248 ),
249 (
250 OutputEvent::RunFinished {
251 generation: 4,
252 attempt: 2,
253 exit_code: Some(0),
254 finished_at: UNIX_EPOCH + Duration::from_millis(2_222),
255 },
256 r#"{"type":"runFinished","generation":4,"attempt":2,"exitCode":0,"finishedAt":2222}"#,
257 ),
258 (
259 OutputEvent::Lagged { skipped: 42 },
260 r#"{"type":"lagged","skipped":42}"#,
261 ),
262 ];
263 for (event, expected) in cases {
264 assert_eq!(serde_json::to_string(&event).unwrap(), expected);
265 }
266 }
267
268 #[test]
269 fn every_event_roundtrips_through_json() {
270 let cases = [
271 OutputEvent::Chunk(OutputChunk {
272 generation: 2,
273 attempt: 1,
274 stream: StreamKind::Stderr,
275 seq: 0,
276 ts: UNIX_EPOCH + Duration::from_millis(1_700_000_000_000),
277 line: Bytes::from_static(b"warning"),
278 }),
279 OutputEvent::RunStarted {
280 generation: 2,
281 attempt: 1,
282 started_at: UNIX_EPOCH + Duration::from_millis(1_700_000_000_000),
283 },
284 OutputEvent::RunFinished {
285 generation: 2,
286 attempt: 1,
287 exit_code: Some(42),
288 finished_at: UNIX_EPOCH + Duration::from_millis(1_700_000_001_000),
289 },
290 OutputEvent::Lagged { skipped: 7 },
291 ];
292
293 for original in cases {
294 let json = serde_json::to_string(&original).unwrap();
295 let back: OutputEvent = serde_json::from_str(&json).unwrap();
296 assert_eq!(back, original, "roundtrip failed for {json}");
297 }
298 }
299
300 #[test]
301 fn binary_chunk_roundtrips_exactly_as_base64() {
302 let chunk = OutputChunk {
303 generation: 1,
304 attempt: 1,
305 stream: StreamKind::Stdout,
306 seq: 0,
307 ts: UNIX_EPOCH,
308 line: Bytes::from_static(&[b'h', b'i', 0xFF, 0xFE]),
309 };
310
311 let json = serde_json::to_string(&chunk).unwrap();
312 assert!(json.contains(r#""line":"aGn//g==""#), "{json}");
313 let decoded: OutputChunk = serde_json::from_str(&json).unwrap();
314 assert_eq!(decoded, chunk);
315 }
316}