peat_mesh/routing/
packet.rs

1//! Data packet types for hierarchical mesh routing
2//!
3//! Defines the structure and metadata for data flowing through the mesh.
4
5use serde::{Deserialize, Serialize};
6
7/// Direction of data flow in the hierarchy
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
9pub enum DataDirection {
10    /// Data flowing upward toward higher hierarchy levels (telemetry aggregation)
11    Upward,
12
13    /// Data flowing downward toward lower hierarchy levels (command dissemination)
14    Downward,
15
16    /// Data exchanged between peers at the same hierarchy level (coordination)
17    Lateral,
18}
19
20/// Type of data being routed
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
22pub enum DataType {
23    /// Telemetry data from sensors/platforms (flows upward)
24    Telemetry,
25
26    /// Status updates (typically flows upward)
27    Status,
28
29    /// Commands for execution (flows downward)
30    Command,
31
32    /// Configuration updates (flows downward)
33    Configuration,
34
35    /// Coordination messages between lateral peers (same level)
36    Coordination,
37
38    /// Aggregated telemetry (flows upward, higher-level summary)
39    AggregatedTelemetry,
40}
41
42impl DataType {
43    /// Get the typical direction for this data type
44    pub fn typical_direction(&self) -> DataDirection {
45        match self {
46            DataType::Telemetry => DataDirection::Upward,
47            DataType::Status => DataDirection::Upward,
48            DataType::Command => DataDirection::Downward,
49            DataType::Configuration => DataDirection::Downward,
50            DataType::Coordination => DataDirection::Lateral,
51            DataType::AggregatedTelemetry => DataDirection::Upward,
52        }
53    }
54
55    /// Check if this data type requires aggregation
56    pub fn requires_aggregation(&self) -> bool {
57        matches!(self, DataType::Telemetry | DataType::Status)
58    }
59}
60
61/// Data packet metadata for routing decisions
62#[derive(Debug, Clone, Serialize, Deserialize)]
63pub struct DataPacket {
64    /// Unique packet identifier
65    pub packet_id: String,
66
67    /// Node ID of the original sender
68    pub source_node_id: String,
69
70    /// Node ID of the intended final recipient (None = broadcast)
71    pub destination_node_id: Option<String>,
72
73    /// Type of data in this packet
74    pub data_type: DataType,
75
76    /// Direction of data flow
77    pub direction: DataDirection,
78
79    /// Hop count (incremented at each forward)
80    pub hop_count: u16,
81
82    /// Maximum hops before packet is dropped (prevents loops)
83    pub max_hops: u16,
84
85    /// Payload data (opaque to router)
86    pub payload: Vec<u8>,
87}
88
89impl DataPacket {
90    /// Create a new telemetry packet
91    pub fn telemetry(source_node_id: impl Into<String>, payload: Vec<u8>) -> Self {
92        Self {
93            packet_id: uuid::Uuid::new_v4().to_string(),
94            source_node_id: source_node_id.into(),
95            destination_node_id: None, // Telemetry flows to HQ (determined by topology)
96            data_type: DataType::Telemetry,
97            direction: DataDirection::Upward,
98            hop_count: 0,
99            max_hops: 10, // Reasonable default for typical hierarchy depth
100            payload,
101        }
102    }
103
104    /// Create a new command packet
105    pub fn command(
106        source_node_id: impl Into<String>,
107        destination_node_id: impl Into<String>,
108        payload: Vec<u8>,
109    ) -> Self {
110        Self {
111            packet_id: uuid::Uuid::new_v4().to_string(),
112            source_node_id: source_node_id.into(),
113            destination_node_id: Some(destination_node_id.into()),
114            data_type: DataType::Command,
115            direction: DataDirection::Downward,
116            hop_count: 0,
117            max_hops: 10,
118            payload,
119        }
120    }
121
122    /// Create a new status update packet
123    pub fn status(source_node_id: impl Into<String>, payload: Vec<u8>) -> Self {
124        Self {
125            packet_id: uuid::Uuid::new_v4().to_string(),
126            source_node_id: source_node_id.into(),
127            destination_node_id: None,
128            data_type: DataType::Status,
129            direction: DataDirection::Upward,
130            hop_count: 0,
131            max_hops: 10,
132            payload,
133        }
134    }
135
136    /// Create a new coordination packet (lateral)
137    pub fn coordination(
138        source_node_id: impl Into<String>,
139        destination_node_id: impl Into<String>,
140        payload: Vec<u8>,
141    ) -> Self {
142        Self {
143            packet_id: uuid::Uuid::new_v4().to_string(),
144            source_node_id: source_node_id.into(),
145            destination_node_id: Some(destination_node_id.into()),
146            data_type: DataType::Coordination,
147            direction: DataDirection::Lateral,
148            hop_count: 0,
149            max_hops: 3, // Lateral messages shouldn't travel far
150            payload,
151        }
152    }
153
154    /// Create an aggregated telemetry packet
155    pub fn aggregated_telemetry(source_node_id: impl Into<String>, payload: Vec<u8>) -> Self {
156        Self {
157            packet_id: uuid::Uuid::new_v4().to_string(),
158            source_node_id: source_node_id.into(),
159            destination_node_id: None,
160            data_type: DataType::AggregatedTelemetry,
161            direction: DataDirection::Upward,
162            hop_count: 0,
163            max_hops: 10,
164            payload,
165        }
166    }
167
168    /// Increment hop count when forwarding packet
169    ///
170    /// Returns true if packet can be forwarded (not yet at max hops),
171    /// false if packet should be dropped.
172    pub fn increment_hop(&mut self) -> bool {
173        self.hop_count += 1;
174        self.hop_count < self.max_hops
175    }
176
177    /// Check if packet has reached maximum hop count
178    pub fn at_max_hops(&self) -> bool {
179        self.hop_count >= self.max_hops
180    }
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186
187    #[test]
188    fn test_telemetry_packet_creation() {
189        let packet = DataPacket::telemetry("node-1", vec![1, 2, 3]);
190
191        assert_eq!(packet.source_node_id, "node-1");
192        assert_eq!(packet.data_type, DataType::Telemetry);
193        assert_eq!(packet.direction, DataDirection::Upward);
194        assert_eq!(packet.hop_count, 0);
195        assert_eq!(packet.max_hops, 10);
196        assert_eq!(packet.payload, vec![1, 2, 3]);
197        assert!(packet.destination_node_id.is_none());
198    }
199
200    #[test]
201    fn test_command_packet_creation() {
202        let packet = DataPacket::command("hq", "node-1", vec![4, 5, 6]);
203
204        assert_eq!(packet.source_node_id, "hq");
205        assert_eq!(packet.destination_node_id, Some("node-1".to_string()));
206        assert_eq!(packet.data_type, DataType::Command);
207        assert_eq!(packet.direction, DataDirection::Downward);
208    }
209
210    #[test]
211    fn test_hop_increment() {
212        let mut packet = DataPacket::telemetry("node-1", vec![]);
213
214        // Should be able to increment up to max_hops - 1
215        for i in 0..9 {
216            assert!(packet.increment_hop(), "Failed at hop {}", i);
217            assert_eq!(packet.hop_count, i + 1);
218        }
219
220        // At hop 10, should return false (at max)
221        assert!(!packet.increment_hop());
222        assert_eq!(packet.hop_count, 10);
223    }
224
225    #[test]
226    fn test_at_max_hops() {
227        let mut packet = DataPacket::telemetry("node-1", vec![]);
228        assert!(!packet.at_max_hops());
229
230        packet.hop_count = 10;
231        assert!(packet.at_max_hops());
232    }
233
234    #[test]
235    fn test_data_type_typical_direction() {
236        assert_eq!(
237            DataType::Telemetry.typical_direction(),
238            DataDirection::Upward
239        );
240        assert_eq!(DataType::Status.typical_direction(), DataDirection::Upward);
241        assert_eq!(
242            DataType::Command.typical_direction(),
243            DataDirection::Downward
244        );
245        assert_eq!(
246            DataType::Configuration.typical_direction(),
247            DataDirection::Downward
248        );
249        assert_eq!(
250            DataType::Coordination.typical_direction(),
251            DataDirection::Lateral
252        );
253        assert_eq!(
254            DataType::AggregatedTelemetry.typical_direction(),
255            DataDirection::Upward
256        );
257    }
258
259    #[test]
260    fn test_data_type_requires_aggregation() {
261        assert!(DataType::Telemetry.requires_aggregation());
262        assert!(DataType::Status.requires_aggregation());
263        assert!(!DataType::Command.requires_aggregation());
264        assert!(!DataType::Configuration.requires_aggregation());
265        assert!(!DataType::Coordination.requires_aggregation());
266        assert!(!DataType::AggregatedTelemetry.requires_aggregation()); // Already aggregated
267    }
268
269    #[test]
270    fn test_status_packet_creation() {
271        let packet = DataPacket::status("node-1", vec![10, 20]);
272
273        assert_eq!(packet.source_node_id, "node-1");
274        assert_eq!(packet.data_type, DataType::Status);
275        assert_eq!(packet.direction, DataDirection::Upward);
276        assert!(packet.destination_node_id.is_none());
277        assert_eq!(packet.hop_count, 0);
278        assert_eq!(packet.max_hops, 10);
279        assert_eq!(packet.payload, vec![10, 20]);
280    }
281
282    #[test]
283    fn test_coordination_packet_creation() {
284        let packet = DataPacket::coordination("node-1", "node-2", vec![5]);
285
286        assert_eq!(packet.source_node_id, "node-1");
287        assert_eq!(packet.destination_node_id, Some("node-2".to_string()));
288        assert_eq!(packet.data_type, DataType::Coordination);
289        assert_eq!(packet.direction, DataDirection::Lateral);
290        assert_eq!(packet.max_hops, 3); // Lateral messages have lower max hops
291    }
292
293    #[test]
294    fn test_aggregated_telemetry_packet_creation() {
295        let packet = DataPacket::aggregated_telemetry("leader-1", vec![99]);
296
297        assert_eq!(packet.source_node_id, "leader-1");
298        assert_eq!(packet.data_type, DataType::AggregatedTelemetry);
299        assert_eq!(packet.direction, DataDirection::Upward);
300        assert!(packet.destination_node_id.is_none());
301        assert_eq!(packet.max_hops, 10);
302    }
303
304    #[test]
305    fn test_increment_hop_returns_false_at_max() {
306        let mut packet = DataPacket::coordination("a", "b", vec![]);
307        // max_hops is 3 for coordination
308        assert!(packet.increment_hop()); // hop_count = 1
309        assert!(packet.increment_hop()); // hop_count = 2
310        assert!(!packet.increment_hop()); // hop_count = 3, equals max_hops
311
312        assert!(packet.at_max_hops());
313    }
314
315    #[test]
316    fn test_at_max_hops_initially_false() {
317        let packet = DataPacket::telemetry("node-1", vec![]);
318        assert!(!packet.at_max_hops());
319    }
320
321    #[test]
322    fn test_packet_unique_ids() {
323        let p1 = DataPacket::telemetry("node-1", vec![1]);
324        let p2 = DataPacket::telemetry("node-1", vec![1]);
325
326        // Each packet should get a unique ID
327        assert_ne!(p1.packet_id, p2.packet_id);
328    }
329
330    #[test]
331    fn test_packet_serialization() {
332        let packet = DataPacket::command("hq", "node-1", vec![1, 2, 3]);
333        let json = serde_json::to_string(&packet).unwrap();
334        let deserialized: DataPacket = serde_json::from_str(&json).unwrap();
335
336        assert_eq!(deserialized.source_node_id, "hq");
337        assert_eq!(deserialized.destination_node_id, Some("node-1".to_string()));
338        assert_eq!(deserialized.data_type, DataType::Command);
339        assert_eq!(deserialized.direction, DataDirection::Downward);
340        assert_eq!(deserialized.payload, vec![1, 2, 3]);
341    }
342
343    #[test]
344    fn test_data_direction_serialization() {
345        let json = serde_json::to_string(&DataDirection::Lateral).unwrap();
346        let deserialized: DataDirection = serde_json::from_str(&json).unwrap();
347        assert_eq!(deserialized, DataDirection::Lateral);
348    }
349}
peat_mesh/routing/packet.rs

peat_mesh/routing/
packet.rs
