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 recognized stream tick clock domain, and normalizes the result
105    /// so the primary `clock_domain` leads and no domain repeats. Ticks on
106    /// other protocol clocks remain attached to the envelope but do not expand
107    /// the stream clock-domain summary. The stored version is always
108    /// [`STREAM_ENVELOPE_VERSION`].
109    #[allow(clippy::too_many_arguments)]
110    pub fn new_with_clock_domains(
111        stream_id: Symbol,
112        packet_id: Symbol,
113        media: StreamMedia,
114        direction: StreamDirection,
115        sequence: u64,
116        ticks: Vec<Tick>,
117        clock_domain: ClockDomain,
118        clock_domains: Vec<ClockDomain>,
119        profile: TransportProfile,
120        diagnostics: Vec<Symbol>,
121        packet: StreamPacket,
122    ) -> Result<Self> {
123        sim_kernel::validate_ticks(&ticks)?;
124        let packet_media = packet.media();
125        if packet_media != media {
126            return Err(Error::Eval(format!(
127                "stream envelope media {} does not match packet media {}",
128                media.symbol(),
129                packet_media.symbol()
130            )));
131        }
132        let mut all_clock_domains = clock_domains;
133        all_clock_domains.extend(
134            ticks
135                .iter()
136                .filter_map(|tick| ClockDomain::from_symbol(&tick.clock).ok()),
137        );
138        let clock_domains = normalize_clock_domains(clock_domain, all_clock_domains);
139        Ok(Self {
140            version: STREAM_ENVELOPE_VERSION,
141            stream_id,
142            packet_id,
143            media,
144            direction,
145            sequence,
146            ticks,
147            clock_domain,
148            clock_domains,
149            profile,
150            diagnostics,
151            packet,
152        })
153    }
154
155    /// Builds an envelope from stream `metadata` and one [`StreamItem`], using
156    /// the in-memory-local [`TransportProfile`].
157    ///
158    /// Derives the packet id from the stream id and `sequence`, and resolves the
159    /// clock domain from the metadata's declared clock. A convenience wrapper
160    /// over [`StreamEnvelope::from_item_with_profile`].
161    pub fn from_item(metadata: &StreamMetadata, sequence: u64, item: &StreamItem) -> Result<Self> {
162        Self::from_item_with_profile(metadata, sequence, item, TransportProfile::memory_local())
163    }
164
165    /// Builds an envelope from stream `metadata` and one [`StreamItem`] under an
166    /// explicit [`TransportProfile`].
167    ///
168    /// Derives the packet id from the stream id and `sequence`, copies the
169    /// item's ticks and packet, and resolves the clock domain from the
170    /// metadata's clock via [`ClockDomain::for_stream_clock`], failing closed
171    /// when the declared clock is unknown. No diagnostics are attached.
172    pub fn from_item_with_profile(
173        metadata: &StreamMetadata,
174        sequence: u64,
175        item: &StreamItem,
176        profile: TransportProfile,
177    ) -> Result<Self> {
178        Self::new(
179            metadata.id().clone(),
180            packet_id(metadata.id(), sequence),
181            metadata.media(),
182            metadata.direction(),
183            sequence,
184            item.ticks().to_vec(),
185            ClockDomain::for_stream_clock(metadata.clock())?,
186            profile,
187            Vec::new(),
188            item.packet().clone(),
189        )
190    }
191
192    /// Returns the envelope wire version (always [`STREAM_ENVELOPE_VERSION`]).
193    pub fn version(&self) -> u32 {
194        self.version
195    }
196
197    /// Returns the id of the stream this envelope belongs to.
198    pub fn stream_id(&self) -> &Symbol {
199        &self.stream_id
200    }
201
202    /// Returns the id of the wrapped packet.
203    pub fn packet_id(&self) -> &Symbol {
204        &self.packet_id
205    }
206
207    /// Returns the media type carried by this envelope.
208    pub fn media(&self) -> StreamMedia {
209        self.media
210    }
211
212    /// Returns the direction of flow for this envelope.
213    pub fn direction(&self) -> StreamDirection {
214        self.direction
215    }
216
217    /// Returns the monotonic sequence number of this envelope within its stream.
218    pub fn sequence(&self) -> u64 {
219        self.sequence
220    }
221
222    /// Returns the [`Tick`]s that locate this envelope on its clocks.
223    pub fn ticks(&self) -> &[Tick] {
224        &self.ticks
225    }
226
227    /// Returns the primary clock domain this envelope is timed against.
228    pub fn clock_domain(&self) -> ClockDomain {
229        self.clock_domain
230    }
231
232    /// Returns every clock domain this envelope rides.
233    ///
234    /// The primary [`ClockDomain::clock_domain`](StreamEnvelope::clock_domain)
235    /// leads, followed by any additional domains contributed by the ticks, with
236    /// no repeats.
237    pub fn clock_domains(&self) -> &[ClockDomain] {
238        &self.clock_domains
239    }
240
241    /// Returns the transport profile bounding what a carrier may do with this
242    /// envelope.
243    pub fn profile(&self) -> &TransportProfile {
244        &self.profile
245    }
246
247    /// Returns the diagnostic symbols attached to this envelope.
248    pub fn diagnostics(&self) -> &[Symbol] {
249        &self.diagnostics
250    }
251
252    /// Returns the wrapped packet payload.
253    pub fn packet(&self) -> &StreamPacket {
254        &self.packet
255    }
256
257    /// Encodes this envelope into its [`Expr`] map wire form.
258    ///
259    /// The map is tagged with [`stream_envelope_tag_symbol`] and round-trips
260    /// back through the [`TryFrom<Expr>`] implementation.
261    pub fn to_expr(&self) -> Expr {
262        Expr::Map(vec![
263            (
264                Expr::Symbol(Symbol::new("envelope")),
265                Expr::Symbol(stream_envelope_tag_symbol()),
266            ),
267            (
268                Expr::Symbol(Symbol::new("version")),
269                Expr::String(self.version.to_string()),
270            ),
271            (
272                Expr::Symbol(Symbol::new("stream-id")),
273                Expr::Symbol(self.stream_id.clone()),
274            ),
275            (
276                Expr::Symbol(Symbol::new("packet-id")),
277                Expr::Symbol(self.packet_id.clone()),
278            ),
279            (
280                Expr::Symbol(Symbol::new("media")),
281                Expr::Symbol(self.media.symbol()),
282            ),
283            (
284                Expr::Symbol(Symbol::new("direction")),
285                Expr::Symbol(self.direction.symbol()),
286            ),
287            (
288                Expr::Symbol(Symbol::new("sequence")),
289                Expr::String(self.sequence.to_string()),
290            ),
291            (
292                Expr::Symbol(Symbol::new("ticks")),
293                Expr::List(self.ticks.iter().map(tick_expr).collect()),
294            ),
295            (
296                Expr::Symbol(Symbol::new("clock-domain")),
297                Expr::Symbol(self.clock_domain.symbol()),
298            ),
299            (
300                Expr::Symbol(Symbol::new("clock-domains")),
301                Expr::List(
302                    self.clock_domains
303                        .iter()
304                        .map(|domain| Expr::Symbol(domain.symbol()))
305                        .collect(),
306                ),
307            ),
308            (Expr::Symbol(Symbol::new("profile")), self.profile.to_expr()),
309            (
310                Expr::Symbol(Symbol::new("diagnostics")),
311                Expr::List(self.diagnostics.iter().cloned().map(Expr::Symbol).collect()),
312            ),
313            (Expr::Symbol(Symbol::new("packet")), self.packet.to_expr()),
314        ])
315    }
316}
317
318impl TryFrom<Expr> for StreamEnvelope {
319    type Error = Error;
320
321    fn try_from(expr: Expr) -> Result<Self> {
322        let Expr::Map(entries) = &expr else {
323            return Err(Error::TypeMismatch {
324                expected: "stream envelope map",
325                found: expr_kind(&expr),
326            });
327        };
328        ensure_fields(
329            entries,
330            &[
331                "envelope",
332                "version",
333                "stream-id",
334                "packet-id",
335                "media",
336                "direction",
337                "sequence",
338                "ticks",
339                "clock-domain",
340                "clock-domains",
341                "profile",
342                "diagnostics",
343                "packet",
344            ],
345        )?;
346        let tag = symbol_field(entries, "envelope")?;
347        if *tag != stream_envelope_tag_symbol() {
348            return Err(Error::Eval(format!(
349                "unknown stream envelope tag {}",
350                tag.as_qualified_str()
351            )));
352        }
353        let version = parse_string_field::<u32>(entries, "version")?;
354        if version != STREAM_ENVELOPE_VERSION {
355            return Err(Error::Eval(format!(
356                "unsupported stream envelope version {version}"
357            )));
358        }
359        let packet = StreamPacket::try_from(field(entries, "packet")?.clone())?;
360        let ticks = tick_list(entries, "ticks")?;
361        Self::new_with_clock_domains(
362            symbol_field(entries, "stream-id")?.clone(),
363            symbol_field(entries, "packet-id")?.clone(),
364            StreamMedia::from_symbol(symbol_field(entries, "media")?)?,
365            StreamDirection::from_symbol(symbol_field(entries, "direction")?)?,
366            parse_string_field::<u64>(entries, "sequence")?,
367            ticks,
368            ClockDomain::from_symbol(symbol_field(entries, "clock-domain")?)?,
369            clock_domain_list(entries, "clock-domains")?,
370            TransportProfile::from_expr(field(entries, "profile")?)?,
371            symbol_list(entries, "diagnostics")?.to_vec(),
372            packet,
373        )
374    }
375}
376
377/// Returns the runtime tag [`Symbol`] that marks a map as a stream envelope.
378///
379/// Written under the `envelope` key by [`StreamEnvelope::to_expr`] and required
380/// on decode by the [`TryFrom<Expr>`] implementation.
381pub fn stream_envelope_tag_symbol() -> Symbol {
382    Symbol::qualified("stream/envelope", "v1")
383}
384
385fn packet_id(stream_id: &Symbol, sequence: u64) -> Symbol {
386    Symbol::qualified(
387        "stream/packet-id",
388        format!("{}#{sequence}", stream_id.as_qualified_str()),
389    )
390}
391
392fn tick_expr(tick: &Tick) -> Expr {
393    Expr::Map(vec![
394        (
395            Expr::Symbol(Symbol::new("clock")),
396            Expr::Symbol(tick.clock.clone()),
397        ),
398        (Expr::Symbol(Symbol::new("index")), ref_expr(&tick.index)),
399    ])
400}
401
402fn tick_from_expr(expr: &Expr) -> Result<Tick> {
403    let Expr::Map(entries) = expr else {
404        return Err(Error::TypeMismatch {
405            expected: "stream tick map",
406            found: expr_kind(expr),
407        });
408    };
409    ensure_fields(entries, &["clock", "index"])?;
410    Ok(Tick::new(
411        symbol_field(entries, "clock")?.clone(),
412        ref_from_expr(field(entries, "index")?)?,
413    ))
414}
415
416fn parse_string_field<T>(entries: &[(Expr, Expr)], name: &str) -> Result<T>
417where
418    T: FromStr,
419    T::Err: std::fmt::Display,
420{
421    string_field(entries, name)?
422        .parse::<T>()
423        .map_err(|err| Error::Eval(format!("invalid stream envelope {name}: {err}")))
424}
425
426fn tick_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Tick>> {
427    list_field(entries, name)?
428        .iter()
429        .map(tick_from_expr)
430        .collect()
431}
432
433fn clock_domain_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<ClockDomain>> {
434    symbol_list(entries, name)?
435        .iter()
436        .map(ClockDomain::from_symbol)
437        .collect()
438}
439
440fn symbol_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Symbol>> {
441    list_field(entries, name)?
442        .iter()
443        .map(|expr| match expr {
444            Expr::Symbol(symbol) => Ok(symbol.clone()),
445            other => Err(Error::TypeMismatch {
446                expected: "symbol list item",
447                found: expr_kind(other),
448            }),
449        })
450        .collect()
451}
452
453fn list_field<'a>(entries: &'a [(Expr, Expr)], name: &str) -> Result<&'a [Expr]> {
454    access::entry_required_list(entries, name, "list field")
455}
456
457fn ensure_fields(entries: &[(Expr, Expr)], allowed: &[&str]) -> Result<()> {
458    for (key, _) in entries {
459        let Expr::Symbol(symbol) = key else {
460            return Err(Error::TypeMismatch {
461                expected: "symbol stream envelope field",
462                found: expr_kind(key),
463            });
464        };
465        if symbol.namespace.is_none() && allowed.contains(&symbol.name.as_ref()) {
466            continue;
467        }
468        return Err(Error::Eval(format!(
469            "unknown stream envelope field {}",
470            symbol.as_qualified_str()
471        )));
472    }
473    Ok(())
474}
475
476fn normalize_clock_domains(
477    primary: ClockDomain,
478    clock_domains: Vec<ClockDomain>,
479) -> Vec<ClockDomain> {
480    let mut domains = vec![primary];
481    for domain in clock_domains {
482        if !domains.contains(&domain) {
483            domains.push(domain);
484        }
485    }
486    domains
487}
sim_lib_stream_core/envelope.rs

sim_lib_stream_core/
envelope.rs
