Skip to main content

dora_message/
metadata.rs

1use std::collections::BTreeMap;
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5
6/// Additional data that is sent as part of output messages.
7///
8/// Includes a timestamp and additional user-provided parameters. The payload is
9/// a self-describing Arrow IPC stream, so the message carries no separate type
10/// descriptor.
11#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
12pub struct Metadata {
13    metadata_version: u16,
14    timestamp: uhlc::Timestamp,
15    pub parameters: MetadataParameters,
16}
17
18impl Metadata {
19    /// Current metadata wire-format version, stamped on every outgoing message.
20    ///
21    /// Bumped from 0 to 1 when the `ArrowTypeInfo` sidecar was dropped and the
22    /// wire format became Arrow-IPC-only, and from 1 to 2 when the binary
23    /// encoding moved from bincode to postcard (varint integers and length
24    /// prefixes, so the byte layout differs even though the field order does
25    /// not). A receiver can compare
26    /// [`metadata_version`](Self::metadata_version) against this to detect a peer
27    /// speaking an incompatible format and report it clearly instead of failing
28    /// with a cryptic positional-deserialization error.
29    pub const CURRENT_VERSION: u16 = 2;
30
31    /// Create metadata with the given timestamp and no user parameters.
32    pub fn new(timestamp: uhlc::Timestamp) -> Self {
33        Self::from_parameters(timestamp, Default::default())
34    }
35
36    /// Metadata for a startup route-probe marker (see [`STARTUP_MARKER_PARAM`]).
37    pub fn startup_marker(timestamp: uhlc::Timestamp) -> Self {
38        Self::from_parameters(
39            timestamp,
40            BTreeMap::from([(STARTUP_MARKER_PARAM.to_owned(), Parameter::Bool(true))]),
41        )
42    }
43
44    /// Whether this message is a startup route-probe marker rather than node
45    /// data. Markers are consumed by the receiving node's startup barrier and
46    /// must never be decoded or surfaced to user code — see
47    /// [`STARTUP_MARKER_PARAM`].
48    pub fn is_startup_marker(&self) -> bool {
49        get_bool_param(&self.parameters, STARTUP_MARKER_PARAM).unwrap_or(false)
50    }
51
52    /// Metadata for a startup route-probe **ack**: the consumer-side reply to a
53    /// startup marker, identifying which consumer input received it (see
54    /// [`STARTUP_ACK_PARAM`]).
55    pub fn startup_ack(timestamp: uhlc::Timestamp, consumer_node: &str, input_id: &str) -> Self {
56        Self::from_parameters(
57            timestamp,
58            BTreeMap::from([
59                (STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(true)),
60                (
61                    STARTUP_ACK_CONSUMER_PARAM.to_owned(),
62                    Parameter::String(consumer_node.to_owned()),
63                ),
64                (
65                    STARTUP_ACK_INPUT_PARAM.to_owned(),
66                    Parameter::String(input_id.to_owned()),
67                ),
68            ]),
69        )
70    }
71
72    /// `Some((consumer_node, input_id))` iff this message is a well-formed
73    /// startup route-probe ack — see [`STARTUP_ACK_PARAM`]. Malformed acks
74    /// (missing or wrongly-typed identity parameters) return `None` and are
75    /// ignored by producers, which keeps the affected output on the reliable
76    /// daemon path instead of switching on bad evidence.
77    pub fn startup_ack_identity(&self) -> Option<(&str, &str)> {
78        if !get_bool_param(&self.parameters, STARTUP_ACK_PARAM).unwrap_or(false) {
79            return None;
80        }
81        let consumer = get_string_param(&self.parameters, STARTUP_ACK_CONSUMER_PARAM)?;
82        let input = get_string_param(&self.parameters, STARTUP_ACK_INPUT_PARAM)?;
83        Some((consumer, input))
84    }
85
86    /// Create metadata with the given timestamp and user parameters, stamping
87    /// the current wire-format version ([`CURRENT_VERSION`](Self::CURRENT_VERSION)).
88    pub fn from_parameters(timestamp: uhlc::Timestamp, parameters: MetadataParameters) -> Self {
89        Self {
90            metadata_version: Self::CURRENT_VERSION,
91            timestamp,
92            parameters,
93        }
94    }
95
96    /// The wire-format version stamped on this metadata. Compare against
97    /// [`CURRENT_VERSION`](Self::CURRENT_VERSION) on receive to reject peers
98    /// using an incompatible format.
99    pub fn metadata_version(&self) -> u16 {
100        self.metadata_version
101    }
102
103    /// The hybrid-logical-clock timestamp assigned when this message was sent.
104    pub fn timestamp(&self) -> uhlc::Timestamp {
105        self.timestamp
106    }
107
108    /// The serialized OpenTelemetry propagation context carried in the
109    /// `open_telemetry_context` parameter, or an empty string if absent.
110    pub fn open_telemetry_context(&self) -> String {
111        get_string_param(&self.parameters, OPEN_TELEMETRY_CONTEXT)
112            .unwrap_or("")
113            .to_string()
114    }
115}
116
117/// Reserved [`MetadataParameters`] key marking a message as a **startup
118/// route-probe marker** rather than node data.
119///
120/// The zenoh data plane is direct node-to-node pub/sub, so a producer that
121/// publishes before a consumer's subscription has propagated would drop those
122/// early samples. Rather than infer route-readiness from zenoh declarations,
123/// each producer publishes markers on its real output topic while an output is
124/// still on the reliable daemon path, and each consumer answers every received
125/// marker with an ack (see [`STARTUP_ACK_PARAM`]): an ack arriving back at the
126/// producer is end-to-end proof that the route pair carries data. The producer
127/// stops marking an output — and switches it to the direct zenoh path — once
128/// all its required consumers have acked, but only within a bounded startup
129/// window: an output still un-acked when the window closes is pinned to the
130/// daemon path for the rest of the run, so a topic's messages never straddle a
131/// path switch (dora-rs/dora#2891). A route that never proves itself just keeps
132/// the output on the daemon path.
133///
134/// The `__dora_` prefix is reserved; user parameters must not use it. Receivers
135/// filter markers before decoding the payload, so they never reach user code.
136pub const STARTUP_MARKER_PARAM: &str = "__dora_startup_marker";
137
138/// Reserved [`MetadataParameters`] key marking a message as a **startup
139/// route-probe ack**: the consumer-side half of the startup handshake.
140///
141/// When a consumer's data subscriber receives a startup marker (see
142/// [`STARTUP_MARKER_PARAM`]) for an input, it replies with an ack on the
143/// output's dedicated `@ack` topic. An ack arriving back at the producer is
144/// end-to-end proof that the route works in *both* directions; once every
145/// required consumer of an output has acked, the producer switches that output
146/// from the reliable daemon path to the direct node-to-node zenoh path. Ack
147/// timing is load-bearing: the switch can only happen inside the producer's
148/// bounded startup window (see [`STARTUP_MARKER_PARAM`]), so a consumer that
149/// acks late costs that output the fast path for the whole run. A missing or
150/// late ack never fails anything — the output simply stays on the daemon path.
151///
152/// Acks travel as an empty-payload message whose attachment carries this flag
153/// plus the acking consumer's identity under [`STARTUP_ACK_CONSUMER_PARAM`] and
154/// [`STARTUP_ACK_INPUT_PARAM`] — the identity rides in the attachment rather
155/// than the zenoh key so consumer/input ids never need key escaping.
156pub const STARTUP_ACK_PARAM: &str = "__dora_startup_ack";
157
158/// Reserved key carrying the acking consumer's node id as a
159/// [`Parameter::String`] — see [`STARTUP_ACK_PARAM`].
160pub const STARTUP_ACK_CONSUMER_PARAM: &str = "__dora_startup_ack_consumer";
161
162/// Reserved key carrying the acking consumer's input id as a
163/// [`Parameter::String`] — see [`STARTUP_ACK_PARAM`].
164pub const STARTUP_ACK_INPUT_PARAM: &str = "__dora_startup_ack_input";
165
166/// Additional metadata that can be sent as part of output messages.
167pub type MetadataParameters = BTreeMap<String, Parameter>;
168
169/// A typed metadata parameter sent as part of output messages.
170///
171/// Parameters are stored by key in [`MetadataParameters`]. The `get_*_param`
172/// helpers ([`get_string_param`], [`get_integer_param`], [`get_bool_param`])
173/// are type-checked: they return the value only when the stored variant matches
174/// the requested type, and `None` both for a missing key **and** for a key whose
175/// stored value has a different type. Callers therefore never need to match on
176/// the variant themselves for the common scalar cases.
177///
178/// ```
179/// use dora_message::metadata::{
180///     get_integer_param, get_string_param, MetadataParameters, Parameter,
181/// };
182///
183/// let mut params = MetadataParameters::new();
184/// params.insert("frame".to_string(), Parameter::Integer(7));
185///
186/// // Matching type -> the value.
187/// assert_eq!(get_integer_param(&params, "frame"), Some(7));
188/// // Wrong requested type -> None (not a panic, not a coercion).
189/// assert_eq!(get_string_param(&params, "frame"), None);
190/// // Missing key -> None.
191/// assert_eq!(get_integer_param(&params, "absent"), None);
192/// ```
193#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
194pub enum Parameter {
195    /// A boolean value.
196    Bool(bool),
197    /// A signed 64-bit integer value.
198    Integer(i64),
199    /// A UTF-8 string value.
200    String(String),
201    /// A list of signed 64-bit integers.
202    ListInt(Vec<i64>),
203    /// A 64-bit floating-point value.
204    Float(f64),
205    /// A list of 64-bit floating-point values.
206    ListFloat(Vec<f64>),
207    /// A list of UTF-8 strings.
208    ListString(Vec<String>),
209    /// A UTC timestamp.
210    Timestamp(DateTime<Utc>),
211}
212
213/// Extract a string parameter from metadata, returning `None` if missing or
214/// not a `Parameter::String`.
215pub fn get_string_param<'a>(params: &'a MetadataParameters, key: &str) -> Option<&'a str> {
216    params.get(key).and_then(|p| match p {
217        Parameter::String(s) => Some(s.as_str()),
218        _ => None,
219    })
220}
221
222/// Extract an integer parameter from metadata, returning `None` if missing or
223/// not a `Parameter::Integer`.
224pub fn get_integer_param(params: &MetadataParameters, key: &str) -> Option<i64> {
225    params.get(key).and_then(|p| match p {
226        Parameter::Integer(n) => Some(*n),
227        _ => None,
228    })
229}
230
231/// Extract a bool parameter from metadata, returning `None` if missing or
232/// not a `Parameter::Bool`.
233pub fn get_bool_param(params: &MetadataParameters, key: &str) -> Option<bool> {
234    params.get(key).and_then(|p| match p {
235        Parameter::Bool(b) => Some(*b),
236        _ => None,
237    })
238}
239
240// ---------------------------------------------------------------------------
241// Well-known metadata parameter keys for service and action patterns
242// ---------------------------------------------------------------------------
243
244/// Metadata key for correlating a service request with its response.
245pub const REQUEST_ID: &str = "request_id";
246
247/// Metadata key for identifying an action goal across feedback/result messages.
248pub const GOAL_ID: &str = "goal_id";
249
250/// Metadata key for the completion status of an action goal.
251pub const GOAL_STATUS: &str = "goal_status";
252
253/// Goal completed successfully.
254pub const GOAL_STATUS_SUCCEEDED: &str = "succeeded";
255
256/// Goal was aborted by the server.
257pub const GOAL_STATUS_ABORTED: &str = "aborted";
258
259/// Goal was canceled by the client.
260pub const GOAL_STATUS_CANCELED: &str = "canceled";
261
262// ---------------------------------------------------------------------------
263// Well-known metadata parameter key for distributed tracing
264// ---------------------------------------------------------------------------
265
266/// Metadata key carrying the serialized OpenTelemetry propagation context, so a
267/// trace can be continued across a dora message hop. Read via
268/// [`Metadata::open_telemetry_context`]; stamped by the node and runtime send
269/// paths when tracing is enabled. Keep this the single source of truth so the
270/// write and read sides can never drift.
271pub const OPEN_TELEMETRY_CONTEXT: &str = "open_telemetry_context";
272
273// ---------------------------------------------------------------------------
274// Well-known metadata parameter keys for the streaming pattern
275// ---------------------------------------------------------------------------
276
277/// Metadata key identifying the conversation/session.
278pub const SESSION_ID: &str = "session_id";
279
280/// Metadata key for the logical segment within a session (e.g. one utterance).
281pub const SEGMENT_ID: &str = "segment_id";
282
283/// Metadata key for chunk sequence number within a segment.
284pub const SEQ: &str = "seq";
285
286/// Metadata key marking the last chunk of a segment (`true` on final chunk).
287pub const FIN: &str = "fin";
288
289/// Metadata key to discard older queued messages on this input (`true` to flush).
290pub const FLUSH: &str = "flush";
291
292/// Metadata key indicating the wire framing of the data payload.
293/// When set to `"arrow-ipc"`, the payload is an Arrow IPC stream.
294pub const FRAMING: &str = "_framing";
295
296/// Value for [`FRAMING`] indicating Arrow IPC stream framing.
297pub const FRAMING_ARROW_IPC: &str = "arrow-ipc";
298
299/// Metadata key carrying the FNV-1a hash (as an `i64`) of the Arrow IPC schema
300/// for a zenoh data message. Present on schema-once messages so a receiver can
301/// tell which primed decoder a schema-less batch belongs to and detect schema
302/// changes. Absent on messages a receiver should decode as a standalone full
303/// stream (large/SHM and daemon-path payloads).
304pub const SCHEMA_HASH: &str = "_schema_hash";
305
306/// Metadata key carrying the true on-wire byte size (as an `i64`) of a
307/// `dora topic` debug frame's data sample.
308///
309/// The daemon rebuilds a self-describing Arrow IPC stream for inspection by
310/// prepending the retained schema block to each schema-once batch, so the
311/// rebuilt stream the CLI receives is larger than what actually travelled on
312/// the wire (the schema-less batch — the schema is published only once on the
313/// `@schema` subtopic). To keep `dora topic info`'s bandwidth accounting
314/// accurate, the daemon stamps the original data-sample length here; the CLI
315/// measures this instead of the rebuilt stream length (dora-rs/dora#2584).
316///
317/// Debug/inspection path only — never set on real node→node outputs.
318pub const WIRE_SIZE: &str = "_wire_size";
319
320/// Byte size to charge for a `dora topic` debug frame when accounting bandwidth.
321///
322/// Prefers the daemon-stamped [`WIRE_SIZE`] (the real on-wire data-sample
323/// length): the `data` the CLI receives is a rebuilt self-describing stream
324/// whose schema was re-prepended for inspection, so for a schema-once output
325/// `data.len()` over-reports what actually travelled. Falls back to the buffer
326/// length when the key is absent — an older daemon, or a non-debug frame that
327/// never carried it (dora-rs/dora#2584). A present-but-out-of-range stamp (a
328/// negative `i64` that can't be a byte count) also falls back rather than
329/// silently counting zero. Keep this the single reader of [`WIRE_SIZE`] so the
330/// daemon stamp and the CLI accounting can never disagree on the fallback rule.
331pub fn debug_frame_wire_size(params: &MetadataParameters, data: Option<&[u8]>) -> usize {
332    get_integer_param(params, WIRE_SIZE)
333        .and_then(|n| usize::try_from(n).ok())
334        .or_else(|| data.map(|d| d.len()))
335        .unwrap_or(0)
336}
337
338/// Returns `true` if the given parameters carry any pattern-correlation key
339/// ([`REQUEST_ID`], [`GOAL_ID`], or [`GOAL_STATUS`]).
340///
341/// Messages marked with these keys belong to a service or action pattern where
342/// multiple Arrow schemas can legitimately flow through a single output/input,
343/// distinguished by metadata rather than a fixed Arrow type. Runtime type
344/// checks skip such messages (dora-rs/adora#150), and the schema-once zenoh
345/// optimization excludes them — on the send side (they always travel as full
346/// self-describing streams), on the node receive side (their schemas must not
347/// churn the per-input decoder), and on the daemon's `dora topic` debug path
348/// (same, for its schema cache). Keep this the single definition so those
349/// layers can never disagree on what "pattern-correlated" means.
350pub fn carries_pattern_correlation(params: &MetadataParameters) -> bool {
351    params.contains_key(REQUEST_ID)
352        || params.contains_key(GOAL_ID)
353        || params.contains_key(GOAL_STATUS)
354}
355
356/// Remove internal wire-protocol keys ([`SCHEMA_HASH`], [`FRAMING`]) from a
357/// parameter map. Call this at every wire→user boundary: the keys are
358/// meaningless after decode, and a stale [`SCHEMA_HASH`] forwarded from an
359/// input's metadata into `send_output` parameters (a standard pattern, e.g.
360/// replay) would ride onto outputs that don't overwrite it, making receivers
361/// hash-mismatch and silently drop them (dora-rs/dora#2366 review).
362pub fn strip_internal_parameters(params: &mut MetadataParameters) {
363    params.remove(SCHEMA_HASH);
364    params.remove(FRAMING);
365}
366
367/// FNV-1a-64 hash with a fixed seed (cross-process deterministic). Used to
368/// fingerprint an Arrow IPC schema block so a schema-less batch can be matched
369/// to the schema it was encoded against (see [`SCHEMA_HASH`]). The producer
370/// node, the consumer node, and the daemon's `dora topic` debug path all hash
371/// the same schema-block bytes with this function, so the value must stay
372/// identical across crates — keep this the single source of truth.
373pub fn fnv1a(bytes: &[u8]) -> u64 {
374    let mut hash: u64 = 0xcbf29ce484222325;
375    for b in bytes {
376        hash ^= *b as u64;
377        hash = hash.wrapping_mul(0x100000001b3);
378    }
379    hash
380}
381
382#[cfg(test)]
383mod tests {
384    use super::*;
385
386    #[test]
387    fn fnv1a_matches_standard_vectors() {
388        // Canonical FNV-1a-64 vectors — pin the algorithm so the producer and
389        // consumers (across crates/processes) never disagree on a schema hash.
390        assert_eq!(fnv1a(b""), 0xcbf29ce484222325);
391        assert_eq!(fnv1a(b"a"), 0xaf63dc4c8601ec8c);
392    }
393
394    fn test_timestamp() -> uhlc::Timestamp {
395        uhlc::HLC::default().new_timestamp()
396    }
397
398    #[test]
399    fn startup_marker_is_detected_and_survives_the_wire() {
400        // A marker is recognized only via the reserved parameter, and the flag
401        // must survive postcard round-tripping — it travels as the zenoh
402        // attachment, and a receiver that failed to recognize it would decode
403        // the marker as node data and surface it to user code.
404        let marker = Metadata::startup_marker(test_timestamp());
405        assert!(marker.is_startup_marker());
406
407        let bytes = crate::encode(&marker).expect("serialize");
408        let decoded: Metadata = crate::decode(&bytes).expect("deserialize");
409        assert!(decoded.is_startup_marker());
410    }
411
412    #[test]
413    fn ordinary_metadata_is_not_a_startup_marker() {
414        // Guard the other direction: real data must never be mistaken for a
415        // marker (it would be silently dropped instead of delivered).
416        assert!(!Metadata::new(test_timestamp()).is_startup_marker());
417
418        // Wrong type under the reserved key must not count as a marker.
419        let wrong_type = Metadata::from_parameters(
420            test_timestamp(),
421            BTreeMap::from([(
422                STARTUP_MARKER_PARAM.to_owned(),
423                Parameter::String("true".into()),
424            )]),
425        );
426        assert!(!wrong_type.is_startup_marker());
427
428        // Explicit `false` is not a marker either.
429        let explicit_false = Metadata::from_parameters(
430            test_timestamp(),
431            BTreeMap::from([(STARTUP_MARKER_PARAM.to_owned(), Parameter::Bool(false))]),
432        );
433        assert!(!explicit_false.is_startup_marker());
434    }
435
436    #[test]
437    fn startup_marker_key_is_reserved_and_distinct() {
438        // The `__dora_` prefix keeps it out of the user parameter namespace, and
439        // it must not collide with any well-known protocol key.
440        assert_eq!(STARTUP_MARKER_PARAM, "__dora_startup_marker");
441        assert!(STARTUP_MARKER_PARAM.starts_with("__dora_"));
442        for key in [
443            REQUEST_ID,
444            GOAL_ID,
445            GOAL_STATUS,
446            SESSION_ID,
447            SEGMENT_ID,
448            SEQ,
449            FIN,
450            FLUSH,
451        ] {
452            assert_ne!(STARTUP_MARKER_PARAM, key);
453        }
454    }
455
456    #[test]
457    fn startup_ack_round_trips_and_extracts_identity() {
458        // The ack travels as a postcard attachment on the `@ack` topic; the
459        // producer must recover exactly the (consumer, input) identity it needs
460        // to tick off a required acker.
461        let ack = Metadata::startup_ack(test_timestamp(), "camera-consumer", "image/depth");
462        assert_eq!(
463            ack.startup_ack_identity(),
464            Some(("camera-consumer", "image/depth"))
465        );
466        // An ack is not a marker (and vice versa, checked below): the two
467        // travel on different topics but share the filtering code path.
468        assert!(!ack.is_startup_marker());
469
470        let bytes = crate::encode(&ack).expect("serialize");
471        let decoded: Metadata = crate::decode(&bytes).expect("deserialize");
472        assert_eq!(
473            decoded.startup_ack_identity(),
474            Some(("camera-consumer", "image/depth"))
475        );
476    }
477
478    #[test]
479    fn malformed_startup_acks_are_rejected() {
480        // Ordinary metadata and markers are not acks.
481        assert_eq!(Metadata::new(test_timestamp()).startup_ack_identity(), None);
482        assert_eq!(
483            Metadata::startup_marker(test_timestamp()).startup_ack_identity(),
484            None
485        );
486
487        // Flag present but identity missing → not a valid ack: a producer must
488        // never count an acker it cannot identify.
489        let flag_only = Metadata::from_parameters(
490            test_timestamp(),
491            BTreeMap::from([(STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(true))]),
492        );
493        assert_eq!(flag_only.startup_ack_identity(), None);
494
495        // Wrongly-typed identity parameters are rejected too.
496        let wrong_types = Metadata::from_parameters(
497            test_timestamp(),
498            BTreeMap::from([
499                (STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(true)),
500                (STARTUP_ACK_CONSUMER_PARAM.to_owned(), Parameter::Integer(1)),
501                (STARTUP_ACK_INPUT_PARAM.to_owned(), Parameter::Integer(2)),
502            ]),
503        );
504        assert_eq!(wrong_types.startup_ack_identity(), None);
505
506        // `Bool(false)` under the flag key is not an ack.
507        let explicit_false = Metadata::from_parameters(
508            test_timestamp(),
509            BTreeMap::from([
510                (STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(false)),
511                (
512                    STARTUP_ACK_CONSUMER_PARAM.to_owned(),
513                    Parameter::String("c".into()),
514                ),
515                (
516                    STARTUP_ACK_INPUT_PARAM.to_owned(),
517                    Parameter::String("i".into()),
518                ),
519            ]),
520        );
521        assert_eq!(explicit_false.startup_ack_identity(), None);
522    }
523
524    #[test]
525    fn startup_ack_keys_are_reserved_and_distinct() {
526        let ack_keys = [
527            STARTUP_ACK_PARAM,
528            STARTUP_ACK_CONSUMER_PARAM,
529            STARTUP_ACK_INPUT_PARAM,
530        ];
531        for key in ack_keys {
532            assert!(key.starts_with("__dora_"));
533            assert_ne!(key, STARTUP_MARKER_PARAM);
534        }
535        for (i, a) in ack_keys.iter().enumerate() {
536            for b in &ack_keys[i + 1..] {
537                assert_ne!(a, b);
538            }
539        }
540    }
541
542    #[test]
543    fn well_known_keys_have_stable_values() {
544        // These string values are part of the cross-language protocol
545        // (Python nodes use the same literal strings). Do not change.
546        assert_eq!(REQUEST_ID, "request_id");
547        assert_eq!(GOAL_ID, "goal_id");
548        assert_eq!(GOAL_STATUS, "goal_status");
549        assert_eq!(GOAL_STATUS_SUCCEEDED, "succeeded");
550        assert_eq!(GOAL_STATUS_ABORTED, "aborted");
551        assert_eq!(GOAL_STATUS_CANCELED, "canceled");
552        assert_eq!(SESSION_ID, "session_id");
553        assert_eq!(SEGMENT_ID, "segment_id");
554        assert_eq!(SEQ, "seq");
555        assert_eq!(FIN, "fin");
556        assert_eq!(FLUSH, "flush");
557    }
558
559    #[test]
560    fn well_known_keys_are_distinct() {
561        let keys = [
562            REQUEST_ID,
563            GOAL_ID,
564            GOAL_STATUS,
565            SESSION_ID,
566            SEGMENT_ID,
567            SEQ,
568            FIN,
569            FLUSH,
570        ];
571        for (i, a) in keys.iter().enumerate() {
572            for b in &keys[i + 1..] {
573                assert_ne!(a, b);
574            }
575        }
576    }
577
578    #[test]
579    fn goal_status_values_are_distinct() {
580        let vals = [
581            GOAL_STATUS_SUCCEEDED,
582            GOAL_STATUS_ABORTED,
583            GOAL_STATUS_CANCELED,
584        ];
585        for (i, a) in vals.iter().enumerate() {
586            for b in &vals[i + 1..] {
587                assert_ne!(a, b);
588            }
589        }
590    }
591
592    #[test]
593    fn outgoing_metadata_is_stamped_with_current_version() {
594        // The wire format is positional (postcard) and carries no separate type
595        // descriptor, so `metadata_version` is the only in-band signal of an
596        // incompatible layout. Every constructor must stamp `CURRENT_VERSION`.
597        assert_eq!(Metadata::CURRENT_VERSION, 2);
598        let ts = uhlc::HLC::default().new_timestamp();
599        assert_eq!(
600            Metadata::new(ts).metadata_version(),
601            Metadata::CURRENT_VERSION
602        );
603        assert_eq!(
604            Metadata::from_parameters(ts, Default::default()).metadata_version(),
605            Metadata::CURRENT_VERSION
606        );
607    }
608
609    #[test]
610    fn get_string_param_extracts_string() {
611        let mut params = MetadataParameters::default();
612        params.insert("key".to_string(), Parameter::String("value".to_string()));
613        assert_eq!(get_string_param(&params, "key"), Some("value"));
614        assert_eq!(get_string_param(&params, "missing"), None);
615    }
616
617    #[test]
618    fn get_string_param_returns_none_for_non_string() {
619        let mut params = MetadataParameters::default();
620        params.insert("num".to_string(), Parameter::Integer(42));
621        assert_eq!(get_string_param(&params, "num"), None);
622    }
623
624    #[test]
625    fn get_integer_param_extracts_integer() {
626        let mut params = MetadataParameters::default();
627        params.insert("key".to_string(), Parameter::Integer(42));
628        assert_eq!(get_integer_param(&params, "key"), Some(42));
629        assert_eq!(get_integer_param(&params, "missing"), None);
630    }
631
632    #[test]
633    fn get_integer_param_returns_none_for_non_integer() {
634        let mut params = MetadataParameters::default();
635        params.insert("s".to_string(), Parameter::String("hello".to_string()));
636        assert_eq!(get_integer_param(&params, "s"), None);
637    }
638
639    #[test]
640    fn get_bool_param_extracts_bool() {
641        let mut params = MetadataParameters::default();
642        params.insert("key".to_string(), Parameter::Bool(true));
643        assert_eq!(get_bool_param(&params, "key"), Some(true));
644        assert_eq!(get_bool_param(&params, "missing"), None);
645    }
646
647    #[test]
648    fn get_bool_param_returns_none_for_non_bool() {
649        let mut params = MetadataParameters::default();
650        params.insert("n".to_string(), Parameter::Integer(1));
651        assert_eq!(get_bool_param(&params, "n"), None);
652    }
653
654    #[test]
655    fn debug_frame_wire_size_prefers_stamped_value() {
656        // The stamped size wins over the (larger, schema-inflated) buffer: a
657        // schema-once frame's rebuilt `data` is bigger than what travelled.
658        let mut params = MetadataParameters::default();
659        params.insert(WIRE_SIZE.to_string(), Parameter::Integer(17));
660        assert_eq!(debug_frame_wire_size(&params, Some(&[0u8; 42])), 17);
661    }
662
663    #[test]
664    fn debug_frame_wire_size_falls_back_to_buffer_len() {
665        // No stamp (older daemon / non-debug frame) ⇒ use the buffer length.
666        let params = MetadataParameters::default();
667        assert_eq!(debug_frame_wire_size(&params, Some(&[0u8; 42])), 42);
668    }
669
670    #[test]
671    fn debug_frame_wire_size_falls_back_on_out_of_range_stamp() {
672        // A negative i64 can't be a byte count; fall back rather than count 0.
673        let mut params = MetadataParameters::default();
674        params.insert(WIRE_SIZE.to_string(), Parameter::Integer(-1));
675        assert_eq!(debug_frame_wire_size(&params, Some(&[0u8; 42])), 42);
676    }
677
678    #[test]
679    fn debug_frame_wire_size_zero_without_stamp_or_buffer() {
680        assert_eq!(
681            debug_frame_wire_size(&MetadataParameters::default(), None),
682            0
683        );
684    }
685}