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/profile.rs"]
20mod profile;
21#[path = "envelope/ref_codec.rs"]
22mod ref_codec;
23
24use sim_kernel::{Error, Expr, Result, Symbol, Tick};
25use sim_value::access;
26
27use crate::buffer::{expr_kind, field, string_field, symbol_field};
28use crate::{StreamDirection, StreamItem, StreamMedia, StreamMetadata, StreamPacket};
29pub use profile::{LatencyClass, StreamCapability, TransportProfile};
30use ref_codec::{ref_expr, ref_from_expr};
31
32/// Wire version of the [`StreamEnvelope`] map form.
33///
34/// Encoded into every [`StreamEnvelope::to_expr`] map and checked on decode;
35/// envelopes carrying any other version are rejected.
36pub const STREAM_ENVELOPE_VERSION: u32 = 1;
37
38/// Clock a stream is timed against.
39///
40/// Each variant names one timeline a packet can ride; the kernel defines the
41/// clock-domain contract as [`Symbol`]s, and this enum is the concrete set this
42/// fabric understands. [`ClockDomain::symbol`] maps a variant to its kernel
43/// symbol and [`ClockDomain::from_symbol`] parses it back, accepting the bare
44/// label, the `clock/<label>` form, and the fully qualified
45/// `stream/clock-domain/<label>` form.
46///
47/// # Examples
48///
49/// ```
50/// use sim_lib_stream_core::ClockDomain;
51///
52/// let domain = ClockDomain::Sample;
53/// assert_eq!(domain.wire_label(), "sample");
54/// let parsed = ClockDomain::from_symbol(&domain.symbol()).unwrap();
55/// assert_eq!(parsed, domain);
56/// ```
57#[derive(Clone, Copy, Debug, PartialEq, Eq)]
58pub enum ClockDomain {
59    /// Per-sample audio timeline (the finest audio clock).
60    Sample,
61    /// Per-block processing timeline (one tick per audio block).
62    Block,
63    /// Control-rate timeline for parameter and modulation updates.
64    Control,
65    /// MIDI tick timeline (musical clock pulses).
66    MidiTick,
67    /// Wall-clock (real-world) time.
68    Wall,
69    /// Transport timeline (musical position: bars/beats under play control).
70    Transport,
71    /// Server-side frame timeline.
72    ServerFrame,
73    /// Browser-side frame timeline (client render cadence).
74    BrowserFrame,
75    /// Trace-step timeline for stepped/replayed execution.
76    TraceStep,
77    /// Job timeline keyed to background job progress.
78    Job,
79}
80
81impl ClockDomain {
82    /// Returns the stable wire label for this domain (for example `"sample"`).
83    pub fn wire_label(self) -> &'static str {
84        match self {
85            Self::Sample => "sample",
86            Self::Block => "block",
87            Self::Control => "control",
88            Self::MidiTick => "midi-tick",
89            Self::Wall => "wall",
90            Self::Transport => "transport",
91            Self::ServerFrame => "server-frame",
92            Self::BrowserFrame => "browser-frame",
93            Self::TraceStep => "trace-step",
94            Self::Job => "job",
95        }
96    }
97
98    /// Returns the kernel [`Symbol`] for this domain, namespaced under
99    /// `stream/clock-domain`.
100    pub fn symbol(self) -> Symbol {
101        Symbol::qualified("stream/clock-domain", self.wire_label())
102    }
103
104    /// Parses a [`ClockDomain`] from a kernel [`Symbol`].
105    ///
106    /// Accepts the bare label, the legacy `clock/<label>` form, and the fully
107    /// qualified `stream/clock-domain/<label>` form. Returns an error for any
108    /// unrecognized clock domain.
109    pub fn from_symbol(symbol: &Symbol) -> Result<Self> {
110        match symbol.as_qualified_str().as_str() {
111            "sample" | "clock/sample" | "stream/clock-domain/sample" => Ok(Self::Sample),
112            "block" | "clock/block" | "stream/clock-domain/block" => Ok(Self::Block),
113            "control" | "clock/control" | "stream/clock-domain/control" => Ok(Self::Control),
114            "midi"
115            | "midi-tick"
116            | "clock/midi"
117            | "clock/midi-tick"
118            | "stream/clock-domain/midi-tick" => Ok(Self::MidiTick),
119            "wall" | "clock/wall" | "stream/clock-domain/wall" => Ok(Self::Wall),
120            "transport" | "clock/transport" | "stream/clock-domain/transport" => {
121                Ok(Self::Transport)
122            }
123            "server-frame" | "clock/server-frame" | "stream/clock-domain/server-frame" => {
124                Ok(Self::ServerFrame)
125            }
126            "browser-frame" | "clock/browser-frame" | "stream/clock-domain/browser-frame" => {
127                Ok(Self::BrowserFrame)
128            }
129            "trace-step" | "clock/trace-step" | "stream/clock-domain/trace-step" => {
130                Ok(Self::TraceStep)
131            }
132            "job" | "clock/job" | "stream/clock-domain/job" => Ok(Self::Job),
133            other => Err(Error::Eval(format!("unknown stream clock domain {other}"))),
134        }
135    }
136
137    /// Resolves the clock domain for a stream's declared clock symbol, falling
138    /// back to [`ClockDomain::ServerFrame`] when the symbol is unrecognized.
139    pub fn for_stream_clock(symbol: &Symbol) -> Self {
140        Self::from_symbol(symbol).unwrap_or(Self::ServerFrame)
141    }
142}
143
144/// A single packet plus the routing and timing metadata that carries it across
145/// the streaming fabric.
146///
147/// Every envelope is constructed through a validating constructor that checks
148/// its [`Tick`]s, confirms the declared [`StreamMedia`] matches the wrapped
149/// [`StreamPacket`], folds each tick's clock into the recorded clock-domain set,
150/// and stamps the current [`STREAM_ENVELOPE_VERSION`]. The struct fields are
151/// private; read access is through the accessor methods, and the wire form is
152/// produced by [`StreamEnvelope::to_expr`] and recovered by [`TryFrom<Expr>`].
153#[derive(Clone, Debug, PartialEq, Eq)]
154pub struct StreamEnvelope {
155    version: u32,
156    stream_id: Symbol,
157    packet_id: Symbol,
158    media: StreamMedia,
159    direction: StreamDirection,
160    sequence: u64,
161    ticks: Vec<Tick>,
162    clock_domain: ClockDomain,
163    clock_domains: Vec<ClockDomain>,
164    profile: TransportProfile,
165    diagnostics: Vec<Symbol>,
166    packet: StreamPacket,
167}
168
169impl StreamEnvelope {
170    /// Builds an envelope whose clock-domain set is seeded from the single
171    /// primary `clock_domain`.
172    ///
173    /// Validates the ticks and checks that `media` matches the packet's media;
174    /// see [`StreamEnvelope::new_with_clock_domains`] for the full contract.
175    #[allow(clippy::too_many_arguments)]
176    pub fn new(
177        stream_id: Symbol,
178        packet_id: Symbol,
179        media: StreamMedia,
180        direction: StreamDirection,
181        sequence: u64,
182        ticks: Vec<Tick>,
183        clock_domain: ClockDomain,
184        profile: TransportProfile,
185        diagnostics: Vec<Symbol>,
186        packet: StreamPacket,
187    ) -> Result<Self> {
188        Self::new_with_clock_domains(
189            stream_id,
190            packet_id,
191            media,
192            direction,
193            sequence,
194            ticks,
195            clock_domain,
196            vec![clock_domain],
197            profile,
198            diagnostics,
199            packet,
200        )
201    }
202
203    /// Builds an envelope with an explicit set of clock domains.
204    ///
205    /// Validates the ticks via the kernel, requires `media` to equal the
206    /// wrapped packet's media (erroring otherwise), augments `clock_domains`
207    /// with each tick's clock domain, and normalizes the result so the primary
208    /// `clock_domain` leads and no domain repeats. The stored version is always
209    /// [`STREAM_ENVELOPE_VERSION`].
210    #[allow(clippy::too_many_arguments)]
211    pub fn new_with_clock_domains(
212        stream_id: Symbol,
213        packet_id: Symbol,
214        media: StreamMedia,
215        direction: StreamDirection,
216        sequence: u64,
217        ticks: Vec<Tick>,
218        clock_domain: ClockDomain,
219        clock_domains: Vec<ClockDomain>,
220        profile: TransportProfile,
221        diagnostics: Vec<Symbol>,
222        packet: StreamPacket,
223    ) -> Result<Self> {
224        sim_kernel::validate_ticks(&ticks)?;
225        let packet_media = packet.media();
226        if packet_media != media {
227            return Err(Error::Eval(format!(
228                "stream envelope media {} does not match packet media {}",
229                media.symbol(),
230                packet_media.symbol()
231            )));
232        }
233        let mut all_clock_domains = clock_domains;
234        for tick in &ticks {
235            all_clock_domains.push(ClockDomain::from_symbol(&tick.clock)?);
236        }
237        let clock_domains = normalize_clock_domains(clock_domain, all_clock_domains);
238        Ok(Self {
239            version: STREAM_ENVELOPE_VERSION,
240            stream_id,
241            packet_id,
242            media,
243            direction,
244            sequence,
245            ticks,
246            clock_domain,
247            clock_domains,
248            profile,
249            diagnostics,
250            packet,
251        })
252    }
253
254    /// Builds an envelope from stream `metadata` and one [`StreamItem`], using
255    /// the in-memory-local [`TransportProfile`].
256    ///
257    /// Derives the packet id from the stream id and `sequence`, and resolves the
258    /// clock domain from the metadata's declared clock. A convenience wrapper
259    /// over [`StreamEnvelope::from_item_with_profile`].
260    pub fn from_item(metadata: &StreamMetadata, sequence: u64, item: &StreamItem) -> Result<Self> {
261        Self::from_item_with_profile(metadata, sequence, item, TransportProfile::memory_local())
262    }
263
264    /// Builds an envelope from stream `metadata` and one [`StreamItem`] under an
265    /// explicit [`TransportProfile`].
266    ///
267    /// Derives the packet id from the stream id and `sequence`, copies the
268    /// item's ticks and packet, and resolves the clock domain from the
269    /// metadata's clock via [`ClockDomain::for_stream_clock`]. No diagnostics
270    /// are attached.
271    pub fn from_item_with_profile(
272        metadata: &StreamMetadata,
273        sequence: u64,
274        item: &StreamItem,
275        profile: TransportProfile,
276    ) -> Result<Self> {
277        Self::new(
278            metadata.id().clone(),
279            packet_id(metadata.id(), sequence),
280            metadata.media(),
281            metadata.direction(),
282            sequence,
283            item.ticks().to_vec(),
284            ClockDomain::for_stream_clock(metadata.clock()),
285            profile,
286            Vec::new(),
287            item.packet().clone(),
288        )
289    }
290
291    /// Returns the envelope wire version (always [`STREAM_ENVELOPE_VERSION`]).
292    pub fn version(&self) -> u32 {
293        self.version
294    }
295
296    /// Returns the id of the stream this envelope belongs to.
297    pub fn stream_id(&self) -> &Symbol {
298        &self.stream_id
299    }
300
301    /// Returns the id of the wrapped packet.
302    pub fn packet_id(&self) -> &Symbol {
303        &self.packet_id
304    }
305
306    /// Returns the media type carried by this envelope.
307    pub fn media(&self) -> StreamMedia {
308        self.media
309    }
310
311    /// Returns the direction of flow for this envelope.
312    pub fn direction(&self) -> StreamDirection {
313        self.direction
314    }
315
316    /// Returns the monotonic sequence number of this envelope within its stream.
317    pub fn sequence(&self) -> u64 {
318        self.sequence
319    }
320
321    /// Returns the [`Tick`]s that locate this envelope on its clocks.
322    pub fn ticks(&self) -> &[Tick] {
323        &self.ticks
324    }
325
326    /// Returns the primary clock domain this envelope is timed against.
327    pub fn clock_domain(&self) -> ClockDomain {
328        self.clock_domain
329    }
330
331    /// Returns every clock domain this envelope rides.
332    ///
333    /// The primary [`ClockDomain::clock_domain`](StreamEnvelope::clock_domain)
334    /// leads, followed by any additional domains contributed by the ticks, with
335    /// no repeats.
336    pub fn clock_domains(&self) -> &[ClockDomain] {
337        &self.clock_domains
338    }
339
340    /// Returns the transport profile bounding what a carrier may do with this
341    /// envelope.
342    pub fn profile(&self) -> &TransportProfile {
343        &self.profile
344    }
345
346    /// Returns the diagnostic symbols attached to this envelope.
347    pub fn diagnostics(&self) -> &[Symbol] {
348        &self.diagnostics
349    }
350
351    /// Returns the wrapped packet payload.
352    pub fn packet(&self) -> &StreamPacket {
353        &self.packet
354    }
355
356    /// Encodes this envelope into its [`Expr`] map wire form.
357    ///
358    /// The map is tagged with [`stream_envelope_tag_symbol`] and round-trips
359    /// back through the [`TryFrom<Expr>`] implementation.
360    pub fn to_expr(&self) -> Expr {
361        Expr::Map(vec![
362            (
363                Expr::Symbol(Symbol::new("envelope")),
364                Expr::Symbol(stream_envelope_tag_symbol()),
365            ),
366            (
367                Expr::Symbol(Symbol::new("version")),
368                Expr::String(self.version.to_string()),
369            ),
370            (
371                Expr::Symbol(Symbol::new("stream-id")),
372                Expr::Symbol(self.stream_id.clone()),
373            ),
374            (
375                Expr::Symbol(Symbol::new("packet-id")),
376                Expr::Symbol(self.packet_id.clone()),
377            ),
378            (
379                Expr::Symbol(Symbol::new("media")),
380                Expr::Symbol(self.media.symbol()),
381            ),
382            (
383                Expr::Symbol(Symbol::new("direction")),
384                Expr::Symbol(self.direction.symbol()),
385            ),
386            (
387                Expr::Symbol(Symbol::new("sequence")),
388                Expr::String(self.sequence.to_string()),
389            ),
390            (
391                Expr::Symbol(Symbol::new("ticks")),
392                Expr::List(self.ticks.iter().map(tick_expr).collect()),
393            ),
394            (
395                Expr::Symbol(Symbol::new("clock-domain")),
396                Expr::Symbol(self.clock_domain.symbol()),
397            ),
398            (
399                Expr::Symbol(Symbol::new("clock-domains")),
400                Expr::List(
401                    self.clock_domains
402                        .iter()
403                        .map(|domain| Expr::Symbol(domain.symbol()))
404                        .collect(),
405                ),
406            ),
407            (Expr::Symbol(Symbol::new("profile")), self.profile.to_expr()),
408            (
409                Expr::Symbol(Symbol::new("diagnostics")),
410                Expr::List(self.diagnostics.iter().cloned().map(Expr::Symbol).collect()),
411            ),
412            (Expr::Symbol(Symbol::new("packet")), self.packet.to_expr()),
413        ])
414    }
415}
416
417impl TryFrom<Expr> for StreamEnvelope {
418    type Error = Error;
419
420    fn try_from(expr: Expr) -> Result<Self> {
421        let Expr::Map(entries) = &expr else {
422            return Err(Error::TypeMismatch {
423                expected: "stream envelope map",
424                found: expr_kind(&expr),
425            });
426        };
427        ensure_fields(
428            entries,
429            &[
430                "envelope",
431                "version",
432                "stream-id",
433                "packet-id",
434                "media",
435                "direction",
436                "sequence",
437                "ticks",
438                "clock-domain",
439                "clock-domains",
440                "profile",
441                "diagnostics",
442                "packet",
443            ],
444        )?;
445        let tag = symbol_field(entries, "envelope")?;
446        if *tag != stream_envelope_tag_symbol() {
447            return Err(Error::Eval(format!(
448                "unknown stream envelope tag {}",
449                tag.as_qualified_str()
450            )));
451        }
452        let version = parse_string_field::<u32>(entries, "version")?;
453        if version != STREAM_ENVELOPE_VERSION {
454            return Err(Error::Eval(format!(
455                "unsupported stream envelope version {version}"
456            )));
457        }
458        let packet = StreamPacket::try_from(field(entries, "packet")?.clone())?;
459        let ticks = tick_list(entries, "ticks")?;
460        Self::new_with_clock_domains(
461            symbol_field(entries, "stream-id")?.clone(),
462            symbol_field(entries, "packet-id")?.clone(),
463            StreamMedia::from_symbol(symbol_field(entries, "media")?)?,
464            StreamDirection::from_symbol(symbol_field(entries, "direction")?)?,
465            parse_string_field::<u64>(entries, "sequence")?,
466            ticks,
467            ClockDomain::from_symbol(symbol_field(entries, "clock-domain")?)?,
468            clock_domain_list(entries, "clock-domains")?,
469            TransportProfile::from_expr(field(entries, "profile")?)?,
470            symbol_list(entries, "diagnostics")?.to_vec(),
471            packet,
472        )
473    }
474}
475
476/// Returns the runtime tag [`Symbol`] that marks a map as a stream envelope.
477///
478/// Written under the `envelope` key by [`StreamEnvelope::to_expr`] and required
479/// on decode by the [`TryFrom<Expr>`] implementation.
480pub fn stream_envelope_tag_symbol() -> Symbol {
481    Symbol::qualified("stream/envelope", "v1")
482}
483
484fn packet_id(stream_id: &Symbol, sequence: u64) -> Symbol {
485    Symbol::qualified(
486        "stream/packet-id",
487        format!("{}#{sequence}", stream_id.as_qualified_str()),
488    )
489}
490
491fn tick_expr(tick: &Tick) -> Expr {
492    Expr::Map(vec![
493        (
494            Expr::Symbol(Symbol::new("clock")),
495            Expr::Symbol(tick.clock.clone()),
496        ),
497        (Expr::Symbol(Symbol::new("index")), ref_expr(&tick.index)),
498    ])
499}
500
501fn tick_from_expr(expr: &Expr) -> Result<Tick> {
502    let Expr::Map(entries) = expr else {
503        return Err(Error::TypeMismatch {
504            expected: "stream tick map",
505            found: expr_kind(expr),
506        });
507    };
508    ensure_fields(entries, &["clock", "index"])?;
509    Ok(Tick::new(
510        symbol_field(entries, "clock")?.clone(),
511        ref_from_expr(field(entries, "index")?)?,
512    ))
513}
514
515fn parse_string_field<T>(entries: &[(Expr, Expr)], name: &str) -> Result<T>
516where
517    T: FromStr,
518    T::Err: std::fmt::Display,
519{
520    string_field(entries, name)?
521        .parse::<T>()
522        .map_err(|err| Error::Eval(format!("invalid stream envelope {name}: {err}")))
523}
524
525fn tick_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Tick>> {
526    list_field(entries, name)?
527        .iter()
528        .map(tick_from_expr)
529        .collect()
530}
531
532fn clock_domain_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<ClockDomain>> {
533    symbol_list(entries, name)?
534        .iter()
535        .map(ClockDomain::from_symbol)
536        .collect()
537}
538
539fn symbol_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Symbol>> {
540    list_field(entries, name)?
541        .iter()
542        .map(|expr| match expr {
543            Expr::Symbol(symbol) => Ok(symbol.clone()),
544            other => Err(Error::TypeMismatch {
545                expected: "symbol list item",
546                found: expr_kind(other),
547            }),
548        })
549        .collect()
550}
551
552fn list_field<'a>(entries: &'a [(Expr, Expr)], name: &str) -> Result<&'a [Expr]> {
553    access::entry_required_list(entries, name, "list field")
554}
555
556fn ensure_fields(entries: &[(Expr, Expr)], allowed: &[&str]) -> Result<()> {
557    for (key, _) in entries {
558        let Expr::Symbol(symbol) = key else {
559            return Err(Error::TypeMismatch {
560                expected: "symbol stream envelope field",
561                found: expr_kind(key),
562            });
563        };
564        if symbol.namespace.is_none() && allowed.contains(&symbol.name.as_ref()) {
565            continue;
566        }
567        return Err(Error::Eval(format!(
568            "unknown stream envelope field {}",
569            symbol.as_qualified_str()
570        )));
571    }
572    Ok(())
573}
574
575fn normalize_clock_domains(
576    primary: ClockDomain,
577    clock_domains: Vec<ClockDomain>,
578) -> Vec<ClockDomain> {
579    let mut domains = vec![primary];
580    for domain in clock_domains {
581        if !domains.contains(&domain) {
582            domains.push(domain);
583        }
584    }
585    domains
586}
sim_lib_stream_core/envelope.rs

sim_lib_stream_core/
envelope.rs
