phoxal_bus/
metadata.rs

1//! `BusMetadata` - the per-sample attachment (D43c/D62/D1).
2//!
3//! The wire body is the plain MessagePack payload (D62); provenance rides here,
4//! in the Zenoh attachment. Identity (which contract, which version) is not
5//! carried in the envelope at all - it lives in the Zenoh key itself (the
6//! version is folded into `<Body as ContractBody>::TOPIC`, D1), so a receiver's
7//! per-key subscription is the whole fast-reject.
8//!
9//! Provenance is [`ProducerId`] plus a per-producer sequence, and the
10//! production instant is an explicit `Option<`[`TimeWindow`]`>` - a sample that
11//! expresses no robot time carries `None`, never a sentinel. The participant id
12//! rides alongside as a diagnostic label; it is never identity, and no
13//! admissibility decision reads it.
14//!
15//! Receiver-side observation time is deliberately absent: it is process-local
16//! and receiver-specific, so it belongs on [`Observed`](crate::handle::Observed),
17//! never on the wire.
18
19use serde::{Deserialize, Serialize};
20
21use crate::abi::CodecId;
22use crate::identity::ProducerId;
23use crate::time::{RobotInstant, TimeWindow};
24
25const MAX_METADATA_BYTES: usize = 4 * 1024;
26pub(crate) const MAX_SOURCE_PARTICIPANT_BYTES: usize = 512;
27
28/// Per-sample metadata carried in the Zenoh attachment (MessagePack-encoded).
29#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
30pub struct BusMetadata {
31    /// The codec used for the body payload.
32    pub codec: u8,
33    /// The producing process.
34    pub producer: ProducerId,
35    /// This producer's monotonically increasing sample sequence, starting at
36    /// zero for every fresh process.
37    pub sequence: u64,
38    /// When this sample's content was produced in robot time, if it expresses
39    /// robot time at all. Commands and diagnostics carry `None`.
40    pub produced_at: Option<TimeWindow>,
41    /// The producing participant id - a diagnostic label only (never identity,
42    /// never an admissibility input).
43    pub participant: String,
44}
45
46impl BusMetadata {
47    /// Encode to the MessagePack attachment bytes.
48    pub fn encode(&self) -> Vec<u8> {
49        let mut bounded = self.clone();
50        if bounded.participant.len() > MAX_SOURCE_PARTICIPANT_BYTES {
51            bounded.participant = truncate_utf8(&bounded.participant, MAX_SOURCE_PARTICIPANT_BYTES);
52        }
53        let encoded =
54            rmp_serde::to_vec_named(&bounded).expect("BusMetadata is always serializable");
55        debug_assert!(encoded.len() <= MAX_METADATA_BYTES);
56        encoded
57    }
58
59    /// Decode from the MessagePack attachment bytes.
60    pub fn decode(bytes: &[u8]) -> Result<Self, rmp_serde::decode::Error> {
61        if bytes.len() > MAX_METADATA_BYTES {
62            return Err(rmp_serde::decode::Error::Syntax(format!(
63                "BusMetadata exceeds the {MAX_METADATA_BYTES}-byte limit"
64            )));
65        }
66        rmp_serde::from_slice(bytes)
67    }
68
69    /// The codec id, if recognized by this wire ABI.
70    pub fn codec_id(&self) -> Option<CodecId> {
71        CodecId::from_u8(self.codec)
72    }
73
74    /// The production instant when it is exactly known.
75    ///
76    /// A state sample published at a logical step is exact; a measurement
77    /// translated from a device clock generally is not, and a consumer that
78    /// needs an exact instant from one has to say so.
79    pub fn produced_exactly_at(&self) -> Option<RobotInstant> {
80        self.produced_at.and_then(TimeWindow::as_exact)
81    }
82}
83
84fn truncate_utf8(value: &str, max_bytes: usize) -> String {
85    if value.len() <= max_bytes {
86        return value.to_string();
87    }
88    let mut end = max_bytes;
89    while !value.is_char_boundary(end) {
90        end -= 1;
91    }
92    value[..end].to_string()
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use crate::identity::TimelineId;
99
100    fn metadata(produced_at: Option<TimeWindow>) -> BusMetadata {
101        BusMetadata {
102            codec: CodecId::MessagePack.as_u8(),
103            producer: ProducerId::mint(),
104            sequence: 7,
105            produced_at,
106            participant: "unit".to_string(),
107        }
108    }
109
110    #[test]
111    fn absence_of_a_production_instant_round_trips_as_absence() {
112        let original = metadata(None);
113        let decoded = BusMetadata::decode(&original.encode()).unwrap();
114        assert_eq!(decoded, original);
115        assert_eq!(decoded.produced_at, None);
116        assert_eq!(decoded.produced_exactly_at(), None);
117    }
118
119    #[test]
120    fn an_exact_production_instant_round_trips_without_collapsing_a_window() {
121        let timeline = TimelineId::mint();
122        let exact = metadata(Some(TimeWindow::exact(RobotInstant::new(timeline, 42))));
123        let decoded = BusMetadata::decode(&exact.encode()).unwrap();
124        assert_eq!(
125            decoded.produced_exactly_at(),
126            Some(RobotInstant::new(timeline, 42))
127        );
128
129        let window = TimeWindow::bounded(
130            RobotInstant::new(timeline, 40),
131            RobotInstant::new(timeline, 44),
132        )
133        .unwrap();
134        let bounded = BusMetadata::decode(&metadata(Some(window)).encode()).unwrap();
135        assert_eq!(bounded.produced_at, Some(window));
136        assert_eq!(
137            bounded.produced_exactly_at(),
138            None,
139            "a bounded estimate must not present itself as exact"
140        );
141    }
142
143    #[test]
144    fn an_over_long_participant_label_is_truncated_at_a_char_boundary() {
145        let mut long = metadata(None);
146        long.participant = "é".repeat(MAX_SOURCE_PARTICIPANT_BYTES);
147        let decoded = BusMetadata::decode(&long.encode()).unwrap();
148        assert!(decoded.participant.len() <= MAX_SOURCE_PARTICIPANT_BYTES);
149        assert!(decoded.participant.chars().all(|c| c == 'é'));
150    }
151}
phoxal_bus/metadata.rs

phoxal_bus/
metadata.rs
