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(¶ms, "frame"), Some(7));
188/// // Wrong requested type -> None (not a panic, not a coercion).
189/// assert_eq!(get_string_param(¶ms, "frame"), None);
190/// // Missing key -> None.
191/// assert_eq!(get_integer_param(¶ms, "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(¶ms, "key"), Some("value"));
614 assert_eq!(get_string_param(¶ms, "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(¶ms, "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(¶ms, "key"), Some(42));
629 assert_eq!(get_integer_param(¶ms, "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(¶ms, "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(¶ms, "key"), Some(true));
644 assert_eq!(get_bool_param(¶ms, "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(¶ms, "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(¶ms, 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(¶ms, 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(¶ms, 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}