sim_lib_stream_core/
envelope.rs

1//! Stream boundary contract: the [`StreamEnvelope`] that wraps every packet
2//! crossing the streaming fabric.
3//!
4//! An envelope binds a [`StreamPacket`] to the routing and timing metadata a
5//! transport needs to carry it: the originating stream and packet ids, the
6//! media and direction, a monotonic sequence number, the [`Tick`]s that locate
7//! it on its clocks, the [`ClockDomain`]s it rides, the [`TransportProfile`]
8//! that bounds what the transport may do, and any diagnostics raised along the
9//! way.
10//!
11//! The kernel owns the protocol vocabulary referenced here -- [`Symbol`],
12//! [`Expr`], [`Tick`], and the clock/capability contracts. This module supplies
13//! the concrete envelope behavior: construction with validation, the wire form
14//! ([`StreamEnvelope::to_expr`] / [`TryFrom<Expr>`]), and the mapping of
15//! clock-domain symbols to the [`ClockDomain`] enum.
16
17use std::str::FromStr;
18
19#[path = "envelope/domain.rs"]
20mod domain;
21#[path = "envelope/profile.rs"]
22mod profile;
23#[path = "envelope/ref_codec.rs"]
24mod ref_codec;
25
26use sim_kernel::{Error, Expr, Result, Symbol, Tick};
27use sim_value::access;
28
29use crate::buffer::{expr_kind, field, string_field, symbol_field};
30use crate::{StreamDirection, StreamItem, StreamMedia, StreamMetadata, StreamPacket};
31pub use domain::ClockDomain;
32pub use profile::{LatencyClass, StreamCapability, TransportProfile};
33use ref_codec::{ref_expr, ref_from_expr};
34
35/// Wire version of the [`StreamEnvelope`] map form.
36///
37/// Encoded into every [`StreamEnvelope::to_expr`] map and checked on decode;
38/// envelopes carrying any other version are rejected.
39pub const STREAM_ENVELOPE_VERSION: u32 = 1;
40
41/// A single packet plus the routing and timing metadata that carries it across
42/// the streaming fabric.
43///
44/// Every envelope is constructed through a validating constructor that checks
45/// its [`Tick`]s, confirms the declared [`StreamMedia`] matches the wrapped
46/// [`StreamPacket`], folds each tick's clock into the recorded clock-domain set,
47/// and stamps the current [`STREAM_ENVELOPE_VERSION`]. The struct fields are
48/// private; read access is through the accessor methods, and the wire form is
49/// produced by [`StreamEnvelope::to_expr`] and recovered by [`TryFrom<Expr>`].
50#[derive(Clone, Debug, PartialEq, Eq)]
51pub struct StreamEnvelope {
52    version: u32,
53    stream_id: Symbol,
54    packet_id: Symbol,
55    media: StreamMedia,
56    direction: StreamDirection,
57    sequence: u64,
58    ticks: Vec<Tick>,
59    clock_domain: ClockDomain,
60    clock_domains: Vec<ClockDomain>,
61    profile: TransportProfile,
62    diagnostics: Vec<Symbol>,
63    packet: StreamPacket,
64}
65
66impl StreamEnvelope {
67    /// Builds an envelope whose clock-domain set is seeded from the single
68    /// primary `clock_domain`.
69    ///
70    /// Validates the ticks and checks that `media` matches the packet's media;
71    /// see [`StreamEnvelope::new_with_clock_domains`] for the full contract.
72    #[allow(clippy::too_many_arguments)]
73    pub fn new(
74        stream_id: Symbol,
75        packet_id: Symbol,
76        media: StreamMedia,
77        direction: StreamDirection,
78        sequence: u64,
79        ticks: Vec<Tick>,
80        clock_domain: ClockDomain,
81        profile: TransportProfile,
82        diagnostics: Vec<Symbol>,
83        packet: StreamPacket,
84    ) -> Result<Self> {
85        Self::new_with_clock_domains(
86            stream_id,
87            packet_id,
88            media,
89            direction,
90            sequence,
91            ticks,
92            clock_domain,
93            vec![clock_domain],
94            profile,
95            diagnostics,
96            packet,
97        )
98    }
99
100    /// Builds an envelope with an explicit set of clock domains.
101    ///
102    /// Validates the ticks via the kernel, requires `media` to equal the
103    /// wrapped packet's media (erroring otherwise), augments `clock_domains`
104    /// with each tick's clock domain, and normalizes the result so the primary
105    /// `clock_domain` leads and no domain repeats. The stored version is always
106    /// [`STREAM_ENVELOPE_VERSION`].
107    #[allow(clippy::too_many_arguments)]
108    pub fn new_with_clock_domains(
109        stream_id: Symbol,
110        packet_id: Symbol,
111        media: StreamMedia,
112        direction: StreamDirection,
113        sequence: u64,
114        ticks: Vec<Tick>,
115        clock_domain: ClockDomain,
116        clock_domains: Vec<ClockDomain>,
117        profile: TransportProfile,
118        diagnostics: Vec<Symbol>,
119        packet: StreamPacket,
120    ) -> Result<Self> {
121        sim_kernel::validate_ticks(&ticks)?;
122        let packet_media = packet.media();
123        if packet_media != media {
124            return Err(Error::Eval(format!(
125                "stream envelope media {} does not match packet media {}",
126                media.symbol(),
127                packet_media.symbol()
128            )));
129        }
130        let mut all_clock_domains = clock_domains;
131        for tick in &ticks {
132            all_clock_domains.push(ClockDomain::from_symbol(&tick.clock)?);
133        }
134        let clock_domains = normalize_clock_domains(clock_domain, all_clock_domains);
135        Ok(Self {
136            version: STREAM_ENVELOPE_VERSION,
137            stream_id,
138            packet_id,
139            media,
140            direction,
141            sequence,
142            ticks,
143            clock_domain,
144            clock_domains,
145            profile,
146            diagnostics,
147            packet,
148        })
149    }
150
151    /// Builds an envelope from stream `metadata` and one [`StreamItem`], using
152    /// the in-memory-local [`TransportProfile`].
153    ///
154    /// Derives the packet id from the stream id and `sequence`, and resolves the
155    /// clock domain from the metadata's declared clock. A convenience wrapper
156    /// over [`StreamEnvelope::from_item_with_profile`].
157    pub fn from_item(metadata: &StreamMetadata, sequence: u64, item: &StreamItem) -> Result<Self> {
158        Self::from_item_with_profile(metadata, sequence, item, TransportProfile::memory_local())
159    }
160
161    /// Builds an envelope from stream `metadata` and one [`StreamItem`] under an
162    /// explicit [`TransportProfile`].
163    ///
164    /// Derives the packet id from the stream id and `sequence`, copies the
165    /// item's ticks and packet, and resolves the clock domain from the
166    /// metadata's clock via [`ClockDomain::for_stream_clock`], failing closed
167    /// when the declared clock is unknown. No diagnostics are attached.
168    pub fn from_item_with_profile(
169        metadata: &StreamMetadata,
170        sequence: u64,
171        item: &StreamItem,
172        profile: TransportProfile,
173    ) -> Result<Self> {
174        Self::new(
175            metadata.id().clone(),
176            packet_id(metadata.id(), sequence),
177            metadata.media(),
178            metadata.direction(),
179            sequence,
180            item.ticks().to_vec(),
181            ClockDomain::for_stream_clock(metadata.clock())?,
182            profile,
183            Vec::new(),
184            item.packet().clone(),
185        )
186    }
187
188    /// Returns the envelope wire version (always [`STREAM_ENVELOPE_VERSION`]).
189    pub fn version(&self) -> u32 {
190        self.version
191    }
192
193    /// Returns the id of the stream this envelope belongs to.
194    pub fn stream_id(&self) -> &Symbol {
195        &self.stream_id
196    }
197
198    /// Returns the id of the wrapped packet.
199    pub fn packet_id(&self) -> &Symbol {
200        &self.packet_id
201    }
202
203    /// Returns the media type carried by this envelope.
204    pub fn media(&self) -> StreamMedia {
205        self.media
206    }
207
208    /// Returns the direction of flow for this envelope.
209    pub fn direction(&self) -> StreamDirection {
210        self.direction
211    }
212
213    /// Returns the monotonic sequence number of this envelope within its stream.
214    pub fn sequence(&self) -> u64 {
215        self.sequence
216    }
217
218    /// Returns the [`Tick`]s that locate this envelope on its clocks.
219    pub fn ticks(&self) -> &[Tick] {
220        &self.ticks
221    }
222
223    /// Returns the primary clock domain this envelope is timed against.
224    pub fn clock_domain(&self) -> ClockDomain {
225        self.clock_domain
226    }
227
228    /// Returns every clock domain this envelope rides.
229    ///
230    /// The primary [`ClockDomain::clock_domain`](StreamEnvelope::clock_domain)
231    /// leads, followed by any additional domains contributed by the ticks, with
232    /// no repeats.
233    pub fn clock_domains(&self) -> &[ClockDomain] {
234        &self.clock_domains
235    }
236
237    /// Returns the transport profile bounding what a carrier may do with this
238    /// envelope.
239    pub fn profile(&self) -> &TransportProfile {
240        &self.profile
241    }
242
243    /// Returns the diagnostic symbols attached to this envelope.
244    pub fn diagnostics(&self) -> &[Symbol] {
245        &self.diagnostics
246    }
247
248    /// Returns the wrapped packet payload.
249    pub fn packet(&self) -> &StreamPacket {
250        &self.packet
251    }
252
253    /// Encodes this envelope into its [`Expr`] map wire form.
254    ///
255    /// The map is tagged with [`stream_envelope_tag_symbol`] and round-trips
256    /// back through the [`TryFrom<Expr>`] implementation.
257    pub fn to_expr(&self) -> Expr {
258        Expr::Map(vec![
259            (
260                Expr::Symbol(Symbol::new("envelope")),
261                Expr::Symbol(stream_envelope_tag_symbol()),
262            ),
263            (
264                Expr::Symbol(Symbol::new("version")),
265                Expr::String(self.version.to_string()),
266            ),
267            (
268                Expr::Symbol(Symbol::new("stream-id")),
269                Expr::Symbol(self.stream_id.clone()),
270            ),
271            (
272                Expr::Symbol(Symbol::new("packet-id")),
273                Expr::Symbol(self.packet_id.clone()),
274            ),
275            (
276                Expr::Symbol(Symbol::new("media")),
277                Expr::Symbol(self.media.symbol()),
278            ),
279            (
280                Expr::Symbol(Symbol::new("direction")),
281                Expr::Symbol(self.direction.symbol()),
282            ),
283            (
284                Expr::Symbol(Symbol::new("sequence")),
285                Expr::String(self.sequence.to_string()),
286            ),
287            (
288                Expr::Symbol(Symbol::new("ticks")),
289                Expr::List(self.ticks.iter().map(tick_expr).collect()),
290            ),
291            (
292                Expr::Symbol(Symbol::new("clock-domain")),
293                Expr::Symbol(self.clock_domain.symbol()),
294            ),
295            (
296                Expr::Symbol(Symbol::new("clock-domains")),
297                Expr::List(
298                    self.clock_domains
299                        .iter()
300                        .map(|domain| Expr::Symbol(domain.symbol()))
301                        .collect(),
302                ),
303            ),
304            (Expr::Symbol(Symbol::new("profile")), self.profile.to_expr()),
305            (
306                Expr::Symbol(Symbol::new("diagnostics")),
307                Expr::List(self.diagnostics.iter().cloned().map(Expr::Symbol).collect()),
308            ),
309            (Expr::Symbol(Symbol::new("packet")), self.packet.to_expr()),
310        ])
311    }
312}
313
314impl TryFrom<Expr> for StreamEnvelope {
315    type Error = Error;
316
317    fn try_from(expr: Expr) -> Result<Self> {
318        let Expr::Map(entries) = &expr else {
319            return Err(Error::TypeMismatch {
320                expected: "stream envelope map",
321                found: expr_kind(&expr),
322            });
323        };
324        ensure_fields(
325            entries,
326            &[
327                "envelope",
328                "version",
329                "stream-id",
330                "packet-id",
331                "media",
332                "direction",
333                "sequence",
334                "ticks",
335                "clock-domain",
336                "clock-domains",
337                "profile",
338                "diagnostics",
339                "packet",
340            ],
341        )?;
342        let tag = symbol_field(entries, "envelope")?;
343        if *tag != stream_envelope_tag_symbol() {
344            return Err(Error::Eval(format!(
345                "unknown stream envelope tag {}",
346                tag.as_qualified_str()
347            )));
348        }
349        let version = parse_string_field::<u32>(entries, "version")?;
350        if version != STREAM_ENVELOPE_VERSION {
351            return Err(Error::Eval(format!(
352                "unsupported stream envelope version {version}"
353            )));
354        }
355        let packet = StreamPacket::try_from(field(entries, "packet")?.clone())?;
356        let ticks = tick_list(entries, "ticks")?;
357        Self::new_with_clock_domains(
358            symbol_field(entries, "stream-id")?.clone(),
359            symbol_field(entries, "packet-id")?.clone(),
360            StreamMedia::from_symbol(symbol_field(entries, "media")?)?,
361            StreamDirection::from_symbol(symbol_field(entries, "direction")?)?,
362            parse_string_field::<u64>(entries, "sequence")?,
363            ticks,
364            ClockDomain::from_symbol(symbol_field(entries, "clock-domain")?)?,
365            clock_domain_list(entries, "clock-domains")?,
366            TransportProfile::from_expr(field(entries, "profile")?)?,
367            symbol_list(entries, "diagnostics")?.to_vec(),
368            packet,
369        )
370    }
371}
372
373/// Returns the runtime tag [`Symbol`] that marks a map as a stream envelope.
374///
375/// Written under the `envelope` key by [`StreamEnvelope::to_expr`] and required
376/// on decode by the [`TryFrom<Expr>`] implementation.
377pub fn stream_envelope_tag_symbol() -> Symbol {
378    Symbol::qualified("stream/envelope", "v1")
379}
380
381fn packet_id(stream_id: &Symbol, sequence: u64) -> Symbol {
382    Symbol::qualified(
383        "stream/packet-id",
384        format!("{}#{sequence}", stream_id.as_qualified_str()),
385    )
386}
387
388fn tick_expr(tick: &Tick) -> Expr {
389    Expr::Map(vec![
390        (
391            Expr::Symbol(Symbol::new("clock")),
392            Expr::Symbol(tick.clock.clone()),
393        ),
394        (Expr::Symbol(Symbol::new("index")), ref_expr(&tick.index)),
395    ])
396}
397
398fn tick_from_expr(expr: &Expr) -> Result<Tick> {
399    let Expr::Map(entries) = expr else {
400        return Err(Error::TypeMismatch {
401            expected: "stream tick map",
402            found: expr_kind(expr),
403        });
404    };
405    ensure_fields(entries, &["clock", "index"])?;
406    Ok(Tick::new(
407        symbol_field(entries, "clock")?.clone(),
408        ref_from_expr(field(entries, "index")?)?,
409    ))
410}
411
412fn parse_string_field<T>(entries: &[(Expr, Expr)], name: &str) -> Result<T>
413where
414    T: FromStr,
415    T::Err: std::fmt::Display,
416{
417    string_field(entries, name)?
418        .parse::<T>()
419        .map_err(|err| Error::Eval(format!("invalid stream envelope {name}: {err}")))
420}
421
422fn tick_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Tick>> {
423    list_field(entries, name)?
424        .iter()
425        .map(tick_from_expr)
426        .collect()
427}
428
429fn clock_domain_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<ClockDomain>> {
430    symbol_list(entries, name)?
431        .iter()
432        .map(ClockDomain::from_symbol)
433        .collect()
434}
435
436fn symbol_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Symbol>> {
437    list_field(entries, name)?
438        .iter()
439        .map(|expr| match expr {
440            Expr::Symbol(symbol) => Ok(symbol.clone()),
441            other => Err(Error::TypeMismatch {
442                expected: "symbol list item",
443                found: expr_kind(other),
444            }),
445        })
446        .collect()
447}
448
449fn list_field<'a>(entries: &'a [(Expr, Expr)], name: &str) -> Result<&'a [Expr]> {
450    access::entry_required_list(entries, name, "list field")
451}
452
453fn ensure_fields(entries: &[(Expr, Expr)], allowed: &[&str]) -> Result<()> {
454    for (key, _) in entries {
455        let Expr::Symbol(symbol) = key else {
456            return Err(Error::TypeMismatch {
457                expected: "symbol stream envelope field",
458                found: expr_kind(key),
459            });
460        };
461        if symbol.namespace.is_none() && allowed.contains(&symbol.name.as_ref()) {
462            continue;
463        }
464        return Err(Error::Eval(format!(
465            "unknown stream envelope field {}",
466            symbol.as_qualified_str()
467        )));
468    }
469    Ok(())
470}
471
472fn normalize_clock_domains(
473    primary: ClockDomain,
474    clock_domains: Vec<ClockDomain>,
475) -> Vec<ClockDomain> {
476    let mut domains = vec![primary];
477    for domain in clock_domains {
478        if !domains.contains(&domain) {
479            domains.push(domain);
480        }
481    }
482    domains
483}
sim_lib_stream_core/envelope.rs

sim_lib_stream_core/
envelope.rs
