1use super::ProtocolError;
4use crate::NodeAddr;
5use crate::tree::TreeCoordinate;
6use std::fmt;
7
8#[derive(Clone, Copy, Debug, PartialEq, Eq)]
25#[repr(u8)]
26pub enum SessionMessageType {
27 DataPacket = 0x10,
31 SenderReport = 0x11,
33 ReceiverReport = 0x12,
35 PathMtuNotification = 0x13,
37 CoordsWarmup = 0x14,
39 EndpointData = 0x15,
41
42 CoordsRequired = 0x20,
45 PathBroken = 0x21,
47 MtuExceeded = 0x22,
49}
50
51impl SessionMessageType {
52 pub fn from_byte(b: u8) -> Option<Self> {
54 match b {
55 0x10 => Some(SessionMessageType::DataPacket),
56 0x11 => Some(SessionMessageType::SenderReport),
57 0x12 => Some(SessionMessageType::ReceiverReport),
58 0x13 => Some(SessionMessageType::PathMtuNotification),
59 0x14 => Some(SessionMessageType::CoordsWarmup),
60 0x15 => Some(SessionMessageType::EndpointData),
61 0x20 => Some(SessionMessageType::CoordsRequired),
62 0x21 => Some(SessionMessageType::PathBroken),
63 0x22 => Some(SessionMessageType::MtuExceeded),
64 _ => None,
65 }
66 }
67
68 pub fn to_byte(self) -> u8 {
70 self as u8
71 }
72}
73
74impl fmt::Display for SessionMessageType {
75 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
76 let name = match self {
77 SessionMessageType::DataPacket => "DataPacket",
78 SessionMessageType::SenderReport => "SenderReport",
79 SessionMessageType::ReceiverReport => "ReceiverReport",
80 SessionMessageType::PathMtuNotification => "PathMtuNotification",
81 SessionMessageType::CoordsWarmup => "CoordsWarmup",
82 SessionMessageType::EndpointData => "EndpointData",
83 SessionMessageType::CoordsRequired => "CoordsRequired",
84 SessionMessageType::PathBroken => "PathBroken",
85 SessionMessageType::MtuExceeded => "MtuExceeded",
86 };
87 write!(f, "{}", name)
88 }
89}
90
91pub(crate) fn coords_wire_size(coords: &TreeCoordinate) -> usize {
97 2 + coords.entries().len() * 16
98}
99
100pub(crate) fn encode_coords(coords: &TreeCoordinate, buf: &mut Vec<u8>) {
105 let addrs: Vec<&NodeAddr> = coords.node_addrs().collect();
106 let count = addrs.len() as u16;
107 buf.extend_from_slice(&count.to_le_bytes());
108 for addr in addrs {
109 buf.extend_from_slice(addr.as_bytes());
110 }
111}
112
113pub(crate) fn decode_coords(data: &[u8]) -> Result<(TreeCoordinate, usize), ProtocolError> {
117 if data.len() < 2 {
118 return Err(ProtocolError::MessageTooShort {
119 expected: 2,
120 got: data.len(),
121 });
122 }
123 let count = u16::from_le_bytes([data[0], data[1]]) as usize;
124 let needed = 2 + count * 16;
125 if data.len() < needed {
126 return Err(ProtocolError::MessageTooShort {
127 expected: needed,
128 got: data.len(),
129 });
130 }
131 if count == 0 {
132 return Err(ProtocolError::Malformed(
133 "coordinate with zero entries".into(),
134 ));
135 }
136 let mut addrs = Vec::with_capacity(count);
137 for i in 0..count {
138 let offset = 2 + i * 16;
139 let mut bytes = [0u8; 16];
140 bytes.copy_from_slice(&data[offset..offset + 16]);
141 addrs.push(NodeAddr::from_bytes(bytes));
142 }
143 let coord =
144 TreeCoordinate::from_addrs(addrs).map_err(|e| ProtocolError::Malformed(e.to_string()))?;
145 Ok((coord, needed))
146}
147
148pub(crate) fn decode_optional_coords(
152 data: &[u8],
153) -> Result<(Option<TreeCoordinate>, usize), ProtocolError> {
154 if data.len() < 2 {
155 return Err(ProtocolError::MessageTooShort {
156 expected: 2,
157 got: data.len(),
158 });
159 }
160 let count = u16::from_le_bytes([data[0], data[1]]) as usize;
161 let needed = 2 + count * 16;
162 if data.len() < needed {
163 return Err(ProtocolError::MessageTooShort {
164 expected: needed,
165 got: data.len(),
166 });
167 }
168 if count == 0 {
169 return Ok((None, 2));
170 }
171 let mut addrs = Vec::with_capacity(count);
172 for i in 0..count {
173 let offset = 2 + i * 16;
174 let mut bytes = [0u8; 16];
175 bytes.copy_from_slice(&data[offset..offset + 16]);
176 addrs.push(NodeAddr::from_bytes(bytes));
177 }
178 let coord =
179 TreeCoordinate::from_addrs(addrs).map_err(|e| ProtocolError::Malformed(e.to_string()))?;
180 Ok((Some(coord), needed))
181}
182
183fn encode_empty_coords(buf: &mut Vec<u8>) {
185 buf.extend_from_slice(&0u16.to_le_bytes());
186}
187
188#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
194pub struct SessionFlags {
195 pub request_ack: bool,
197 pub bidirectional: bool,
199}
200
201impl SessionFlags {
202 pub fn new() -> Self {
204 Self::default()
205 }
206
207 pub fn with_ack(mut self) -> Self {
209 self.request_ack = true;
210 self
211 }
212
213 pub fn bidirectional(mut self) -> Self {
215 self.bidirectional = true;
216 self
217 }
218
219 pub fn to_byte(&self) -> u8 {
221 let mut flags = 0u8;
222 if self.request_ack {
223 flags |= 0x01;
224 }
225 if self.bidirectional {
226 flags |= 0x02;
227 }
228 flags
229 }
230
231 pub fn from_byte(byte: u8) -> Self {
233 Self {
234 request_ack: byte & 0x01 != 0,
235 bidirectional: byte & 0x02 != 0,
236 }
237 }
238}
239
240#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
253pub struct FspFlags {
254 pub coords_present: bool,
256 pub key_epoch: bool,
258 pub unencrypted: bool,
260}
261
262impl FspFlags {
263 pub fn new() -> Self {
265 Self::default()
266 }
267
268 pub fn to_byte(&self) -> u8 {
270 let mut flags = 0u8;
271 if self.coords_present {
272 flags |= 0x01;
273 }
274 if self.key_epoch {
275 flags |= 0x02;
276 }
277 if self.unencrypted {
278 flags |= 0x04;
279 }
280 flags
281 }
282
283 pub fn from_byte(byte: u8) -> Self {
285 Self {
286 coords_present: byte & 0x01 != 0,
287 key_epoch: byte & 0x02 != 0,
288 unencrypted: byte & 0x04 != 0,
289 }
290 }
291}
292
293#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
300pub struct FspInnerFlags {
301 pub spin_bit: bool,
303}
304
305impl FspInnerFlags {
306 pub fn new() -> Self {
308 Self::default()
309 }
310
311 pub fn to_byte(&self) -> u8 {
313 if self.spin_bit { 0x01 } else { 0x00 }
314 }
315
316 pub fn from_byte(byte: u8) -> Self {
318 Self {
319 spin_bit: byte & 0x01 != 0,
320 }
321 }
322}
323
324#[derive(Clone, Debug)]
356pub struct SessionSetup {
357 pub src_coords: TreeCoordinate,
359 pub dest_coords: TreeCoordinate,
361 pub flags: SessionFlags,
363 pub handshake_payload: Vec<u8>,
365}
366
367impl SessionSetup {
368 pub fn new(src_coords: TreeCoordinate, dest_coords: TreeCoordinate) -> Self {
370 Self {
371 src_coords,
372 dest_coords,
373 flags: SessionFlags::new(),
374 handshake_payload: Vec::new(),
375 }
376 }
377
378 pub fn with_flags(mut self, flags: SessionFlags) -> Self {
380 self.flags = flags;
381 self
382 }
383
384 pub fn with_handshake(mut self, payload: Vec<u8>) -> Self {
386 self.handshake_payload = payload;
387 self
388 }
389
390 pub fn encode(&self) -> Vec<u8> {
395 let mut body = Vec::new();
397 body.push(self.flags.to_byte());
398 encode_coords(&self.src_coords, &mut body);
399 encode_coords(&self.dest_coords, &mut body);
400 let hs_len = self.handshake_payload.len() as u16;
401 body.extend_from_slice(&hs_len.to_le_bytes());
402 body.extend_from_slice(&self.handshake_payload);
403
404 let payload_len = body.len() as u16;
406 let mut buf = Vec::with_capacity(4 + body.len());
407 buf.push(0x01); buf.push(0x00); buf.extend_from_slice(&payload_len.to_le_bytes());
410 buf.extend_from_slice(&body);
411 buf
412 }
413
414 pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
416 if payload.is_empty() {
417 return Err(ProtocolError::MessageTooShort {
418 expected: 1,
419 got: 0,
420 });
421 }
422 let flags = SessionFlags::from_byte(payload[0]);
423 let mut offset = 1;
424
425 let (src_coords, consumed) = decode_coords(&payload[offset..])?;
426 offset += consumed;
427
428 let (dest_coords, consumed) = decode_coords(&payload[offset..])?;
429 offset += consumed;
430
431 if payload.len() < offset + 2 {
432 return Err(ProtocolError::MessageTooShort {
433 expected: offset + 2,
434 got: payload.len(),
435 });
436 }
437 let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize;
438 offset += 2;
439
440 if payload.len() < offset + hs_len {
441 return Err(ProtocolError::MessageTooShort {
442 expected: offset + hs_len,
443 got: payload.len(),
444 });
445 }
446 let handshake_payload = payload[offset..offset + hs_len].to_vec();
447
448 Ok(Self {
449 src_coords,
450 dest_coords,
451 flags,
452 handshake_payload,
453 })
454 }
455}
456
457#[derive(Clone, Debug)]
489pub struct SessionAck {
490 pub src_coords: TreeCoordinate,
492 pub dest_coords: TreeCoordinate,
494 pub flags: u8,
496 pub handshake_payload: Vec<u8>,
498}
499
500impl SessionAck {
501 pub fn new(src_coords: TreeCoordinate, dest_coords: TreeCoordinate) -> Self {
503 Self {
504 src_coords,
505 dest_coords,
506 flags: 0,
507 handshake_payload: Vec::new(),
508 }
509 }
510
511 pub fn with_handshake(mut self, payload: Vec<u8>) -> Self {
513 self.handshake_payload = payload;
514 self
515 }
516
517 pub fn encode(&self) -> Vec<u8> {
522 let mut body = Vec::new();
524 body.push(self.flags);
525 encode_coords(&self.src_coords, &mut body);
526 encode_coords(&self.dest_coords, &mut body);
527 let hs_len = self.handshake_payload.len() as u16;
528 body.extend_from_slice(&hs_len.to_le_bytes());
529 body.extend_from_slice(&self.handshake_payload);
530
531 let payload_len = body.len() as u16;
533 let mut buf = Vec::with_capacity(4 + body.len());
534 buf.push(0x02); buf.push(0x00); buf.extend_from_slice(&payload_len.to_le_bytes());
537 buf.extend_from_slice(&body);
538 buf
539 }
540
541 pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
543 if payload.is_empty() {
544 return Err(ProtocolError::MessageTooShort {
545 expected: 1,
546 got: 0,
547 });
548 }
549 let flags = payload[0];
550 let mut offset = 1;
551
552 let (src_coords, consumed) = decode_coords(&payload[offset..])?;
553 offset += consumed;
554
555 let (dest_coords, consumed) = decode_coords(&payload[offset..])?;
556 offset += consumed;
557
558 if payload.len() < offset + 2 {
559 return Err(ProtocolError::MessageTooShort {
560 expected: offset + 2,
561 got: payload.len(),
562 });
563 }
564 let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize;
565 offset += 2;
566
567 if payload.len() < offset + hs_len {
568 return Err(ProtocolError::MessageTooShort {
569 expected: offset + hs_len,
570 got: payload.len(),
571 });
572 }
573 let handshake_payload = payload[offset..offset + hs_len].to_vec();
574
575 Ok(Self {
576 src_coords,
577 dest_coords,
578 flags,
579 handshake_payload,
580 })
581 }
582}
583
584#[derive(Clone, Debug)]
602pub struct SessionMsg3 {
603 pub flags: u8,
605 pub handshake_payload: Vec<u8>,
607}
608
609impl SessionMsg3 {
610 pub fn new(handshake_payload: Vec<u8>) -> Self {
612 Self {
613 flags: 0,
614 handshake_payload,
615 }
616 }
617
618 pub fn encode(&self) -> Vec<u8> {
623 let mut body = Vec::new();
625 body.push(self.flags);
626 let hs_len = self.handshake_payload.len() as u16;
627 body.extend_from_slice(&hs_len.to_le_bytes());
628 body.extend_from_slice(&self.handshake_payload);
629
630 let payload_len = body.len() as u16;
632 let mut buf = Vec::with_capacity(4 + body.len());
633 buf.push(0x03); buf.push(0x00); buf.extend_from_slice(&payload_len.to_le_bytes());
636 buf.extend_from_slice(&body);
637 buf
638 }
639
640 pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
642 if payload.is_empty() {
643 return Err(ProtocolError::MessageTooShort {
644 expected: 1,
645 got: 0,
646 });
647 }
648 let flags = payload[0];
649 let mut offset = 1;
650
651 if payload.len() < offset + 2 {
652 return Err(ProtocolError::MessageTooShort {
653 expected: offset + 2,
654 got: payload.len(),
655 });
656 }
657 let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize;
658 offset += 2;
659
660 if payload.len() < offset + hs_len {
661 return Err(ProtocolError::MessageTooShort {
662 expected: offset + hs_len,
663 got: payload.len(),
664 });
665 }
666 let handshake_payload = payload[offset..offset + hs_len].to_vec();
667
668 Ok(Self {
669 flags,
670 handshake_payload,
671 })
672 }
673}
674
675#[derive(Debug, Clone, PartialEq, Eq)]
698pub struct SessionSenderReport {
699 pub interval_start_counter: u64,
700 pub interval_end_counter: u64,
701 pub interval_start_timestamp: u32,
702 pub interval_end_timestamp: u32,
703 pub interval_bytes_sent: u32,
704 pub cumulative_packets_sent: u64,
705 pub cumulative_bytes_sent: u64,
706}
707
708pub const SESSION_SENDER_REPORT_SIZE: usize = 46;
710
711impl SessionSenderReport {
712 pub fn encode(&self) -> Vec<u8> {
714 let mut buf = Vec::with_capacity(SESSION_SENDER_REPORT_SIZE);
715 buf.extend_from_slice(&[0u8; 2]); buf.extend_from_slice(&self.interval_start_counter.to_le_bytes());
717 buf.extend_from_slice(&self.interval_end_counter.to_le_bytes());
718 buf.extend_from_slice(&self.interval_start_timestamp.to_le_bytes());
719 buf.extend_from_slice(&self.interval_end_timestamp.to_le_bytes());
720 buf.extend_from_slice(&self.interval_bytes_sent.to_le_bytes());
721 buf.extend_from_slice(&self.cumulative_packets_sent.to_le_bytes());
722 buf.extend_from_slice(&self.cumulative_bytes_sent.to_le_bytes());
723 buf
724 }
725
726 pub fn decode(body: &[u8]) -> Result<Self, ProtocolError> {
728 if body.len() < SESSION_SENDER_REPORT_SIZE {
729 return Err(ProtocolError::MessageTooShort {
730 expected: SESSION_SENDER_REPORT_SIZE,
731 got: body.len(),
732 });
733 }
734 let p = &body[2..];
736 Ok(Self {
737 interval_start_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()),
738 interval_end_counter: u64::from_le_bytes(p[8..16].try_into().unwrap()),
739 interval_start_timestamp: u32::from_le_bytes(p[16..20].try_into().unwrap()),
740 interval_end_timestamp: u32::from_le_bytes(p[20..24].try_into().unwrap()),
741 interval_bytes_sent: u32::from_le_bytes(p[24..28].try_into().unwrap()),
742 cumulative_packets_sent: u64::from_le_bytes(p[28..36].try_into().unwrap()),
743 cumulative_bytes_sent: u64::from_le_bytes(p[36..44].try_into().unwrap()),
744 })
745 }
746}
747
748#[derive(Debug, Clone, PartialEq, Eq)]
774pub struct SessionReceiverReport {
775 pub highest_counter: u64,
776 pub cumulative_packets_recv: u64,
777 pub cumulative_bytes_recv: u64,
778 pub timestamp_echo: u32,
779 pub dwell_time: u16,
780 pub max_burst_loss: u16,
781 pub mean_burst_loss: u16,
782 pub jitter: u32,
783 pub ecn_ce_count: u32,
784 pub owd_trend: i32,
785 pub burst_loss_count: u32,
786 pub cumulative_reorder_count: u32,
787 pub interval_packets_recv: u32,
788 pub interval_bytes_recv: u32,
789}
790
791pub const SESSION_RECEIVER_REPORT_SIZE: usize = 66;
793
794impl SessionReceiverReport {
795 pub fn encode(&self) -> Vec<u8> {
797 let mut buf = Vec::with_capacity(SESSION_RECEIVER_REPORT_SIZE);
798 buf.extend_from_slice(&[0u8; 2]); buf.extend_from_slice(&self.highest_counter.to_le_bytes());
800 buf.extend_from_slice(&self.cumulative_packets_recv.to_le_bytes());
801 buf.extend_from_slice(&self.cumulative_bytes_recv.to_le_bytes());
802 buf.extend_from_slice(&self.timestamp_echo.to_le_bytes());
803 buf.extend_from_slice(&self.dwell_time.to_le_bytes());
804 buf.extend_from_slice(&self.max_burst_loss.to_le_bytes());
805 buf.extend_from_slice(&self.mean_burst_loss.to_le_bytes());
806 buf.extend_from_slice(&[0u8; 2]); buf.extend_from_slice(&self.jitter.to_le_bytes());
808 buf.extend_from_slice(&self.ecn_ce_count.to_le_bytes());
809 buf.extend_from_slice(&self.owd_trend.to_le_bytes());
810 buf.extend_from_slice(&self.burst_loss_count.to_le_bytes());
811 buf.extend_from_slice(&self.cumulative_reorder_count.to_le_bytes());
812 buf.extend_from_slice(&self.interval_packets_recv.to_le_bytes());
813 buf.extend_from_slice(&self.interval_bytes_recv.to_le_bytes());
814 buf
815 }
816
817 pub fn decode(body: &[u8]) -> Result<Self, ProtocolError> {
819 if body.len() < SESSION_RECEIVER_REPORT_SIZE {
820 return Err(ProtocolError::MessageTooShort {
821 expected: SESSION_RECEIVER_REPORT_SIZE,
822 got: body.len(),
823 });
824 }
825 let p = &body[2..];
827 Ok(Self {
828 highest_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()),
829 cumulative_packets_recv: u64::from_le_bytes(p[8..16].try_into().unwrap()),
830 cumulative_bytes_recv: u64::from_le_bytes(p[16..24].try_into().unwrap()),
831 timestamp_echo: u32::from_le_bytes(p[24..28].try_into().unwrap()),
832 dwell_time: u16::from_le_bytes(p[28..30].try_into().unwrap()),
833 max_burst_loss: u16::from_le_bytes(p[30..32].try_into().unwrap()),
834 mean_burst_loss: u16::from_le_bytes(p[32..34].try_into().unwrap()),
835 jitter: u32::from_le_bytes(p[36..40].try_into().unwrap()),
837 ecn_ce_count: u32::from_le_bytes(p[40..44].try_into().unwrap()),
838 owd_trend: i32::from_le_bytes(p[44..48].try_into().unwrap()),
839 burst_loss_count: u32::from_le_bytes(p[48..52].try_into().unwrap()),
840 cumulative_reorder_count: u32::from_le_bytes(p[52..56].try_into().unwrap()),
841 interval_packets_recv: u32::from_le_bytes(p[56..60].try_into().unwrap()),
842 interval_bytes_recv: u32::from_le_bytes(p[60..64].try_into().unwrap()),
843 })
844 }
845}
846
847#[derive(Debug, Clone, PartialEq, Eq)]
859pub struct PathMtuNotification {
860 pub path_mtu: u16,
862}
863
864pub const PATH_MTU_NOTIFICATION_SIZE: usize = 2;
866
867impl PathMtuNotification {
868 pub fn new(path_mtu: u16) -> Self {
870 Self { path_mtu }
871 }
872
873 pub fn encode(&self) -> Vec<u8> {
875 self.path_mtu.to_le_bytes().to_vec()
876 }
877
878 pub fn decode(body: &[u8]) -> Result<Self, ProtocolError> {
880 if body.len() < PATH_MTU_NOTIFICATION_SIZE {
881 return Err(ProtocolError::MessageTooShort {
882 expected: PATH_MTU_NOTIFICATION_SIZE,
883 got: body.len(),
884 });
885 }
886 Ok(Self {
887 path_mtu: u16::from_le_bytes([body[0], body[1]]),
888 })
889 }
890}
891
892#[derive(Clone, Debug)]
916pub struct CoordsRequired {
917 pub dest_addr: NodeAddr,
919 pub reporter: NodeAddr,
921}
922
923pub const COORDS_REQUIRED_SIZE: usize = 34;
925
926impl CoordsRequired {
927 pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self {
929 Self {
930 dest_addr,
931 reporter,
932 }
933 }
934
935 pub fn encode(&self) -> Vec<u8> {
939 let body_len = 1 + 1 + 16 + 16; let mut buf = Vec::with_capacity(4 + body_len);
942 buf.push(0x00); buf.push(0x04); let payload_len = body_len as u16;
946 buf.extend_from_slice(&payload_len.to_le_bytes());
947 buf.push(SessionMessageType::CoordsRequired.to_byte());
949 buf.push(0x00); buf.extend_from_slice(self.dest_addr.as_bytes());
951 buf.extend_from_slice(self.reporter.as_bytes());
952 buf
953 }
954
955 pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
957 if payload.len() < 33 {
959 return Err(ProtocolError::MessageTooShort {
960 expected: 33,
961 got: payload.len(),
962 });
963 }
964 let mut dest_bytes = [0u8; 16];
966 dest_bytes.copy_from_slice(&payload[1..17]);
967 let mut reporter_bytes = [0u8; 16];
968 reporter_bytes.copy_from_slice(&payload[17..33]);
969
970 Ok(Self {
971 dest_addr: NodeAddr::from_bytes(dest_bytes),
972 reporter: NodeAddr::from_bytes(reporter_bytes),
973 })
974 }
975}
976
977#[derive(Clone, Debug)]
995pub struct PathBroken {
996 pub dest_addr: NodeAddr,
998 pub reporter: NodeAddr,
1000 pub last_known_coords: Option<TreeCoordinate>,
1002}
1003
1004impl PathBroken {
1005 pub fn new(dest_addr: NodeAddr, reporter: NodeAddr) -> Self {
1007 Self {
1008 dest_addr,
1009 reporter,
1010 last_known_coords: None,
1011 }
1012 }
1013
1014 pub fn with_last_coords(mut self, coords: TreeCoordinate) -> Self {
1016 self.last_known_coords = Some(coords);
1017 self
1018 }
1019
1020 pub fn encode(&self) -> Vec<u8> {
1024 let mut body = Vec::new();
1026 body.push(SessionMessageType::PathBroken.to_byte());
1027 body.push(0x00); body.extend_from_slice(self.dest_addr.as_bytes());
1029 body.extend_from_slice(self.reporter.as_bytes());
1030 if let Some(ref coords) = self.last_known_coords {
1031 encode_coords(coords, &mut body);
1032 } else {
1033 encode_empty_coords(&mut body);
1034 }
1035
1036 let payload_len = body.len() as u16;
1038 let mut buf = Vec::with_capacity(4 + body.len());
1039 buf.push(0x00); buf.push(0x04); buf.extend_from_slice(&payload_len.to_le_bytes());
1042 buf.extend_from_slice(&body);
1043 buf
1044 }
1045
1046 pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
1048 if payload.len() < 35 {
1050 return Err(ProtocolError::MessageTooShort {
1051 expected: 35,
1052 got: payload.len(),
1053 });
1054 }
1055 let mut dest_bytes = [0u8; 16];
1057 dest_bytes.copy_from_slice(&payload[1..17]);
1058 let mut reporter_bytes = [0u8; 16];
1059 reporter_bytes.copy_from_slice(&payload[17..33]);
1060
1061 let (last_known_coords, _consumed) = decode_optional_coords(&payload[33..])?;
1062
1063 Ok(Self {
1064 dest_addr: NodeAddr::from_bytes(dest_bytes),
1065 reporter: NodeAddr::from_bytes(reporter_bytes),
1066 last_known_coords,
1067 })
1068 }
1069}
1070
1071#[derive(Clone, Debug)]
1089pub struct MtuExceeded {
1090 pub dest_addr: NodeAddr,
1092 pub reporter: NodeAddr,
1094 pub mtu: u16,
1096}
1097
1098pub const MTU_EXCEEDED_SIZE: usize = 36;
1100
1101impl MtuExceeded {
1102 pub fn new(dest_addr: NodeAddr, reporter: NodeAddr, mtu: u16) -> Self {
1104 Self {
1105 dest_addr,
1106 reporter,
1107 mtu,
1108 }
1109 }
1110
1111 pub fn encode(&self) -> Vec<u8> {
1115 let body_len = MTU_EXCEEDED_SIZE; let mut buf = Vec::with_capacity(4 + body_len);
1117 buf.push(0x00); buf.push(0x04); let payload_len = body_len as u16;
1121 buf.extend_from_slice(&payload_len.to_le_bytes());
1122 buf.push(SessionMessageType::MtuExceeded.to_byte());
1124 buf.push(0x00); buf.extend_from_slice(self.dest_addr.as_bytes());
1126 buf.extend_from_slice(self.reporter.as_bytes());
1127 buf.extend_from_slice(&self.mtu.to_le_bytes());
1128 buf
1129 }
1130
1131 pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
1133 if payload.len() < 35 {
1135 return Err(ProtocolError::MessageTooShort {
1136 expected: 35,
1137 got: payload.len(),
1138 });
1139 }
1140 let mut dest_bytes = [0u8; 16];
1142 dest_bytes.copy_from_slice(&payload[1..17]);
1143 let mut reporter_bytes = [0u8; 16];
1144 reporter_bytes.copy_from_slice(&payload[17..33]);
1145 let mtu = u16::from_le_bytes([payload[33], payload[34]]);
1146
1147 Ok(Self {
1148 dest_addr: NodeAddr::from_bytes(dest_bytes),
1149 reporter: NodeAddr::from_bytes(reporter_bytes),
1150 mtu,
1151 })
1152 }
1153}
1154
1155#[cfg(test)]
1156mod tests {
1157 use super::*;
1158
1159 fn make_node_addr(val: u8) -> NodeAddr {
1160 let mut bytes = [0u8; 16];
1161 bytes[0] = val;
1162 NodeAddr::from_bytes(bytes)
1163 }
1164
1165 fn make_coords(ids: &[u8]) -> TreeCoordinate {
1166 TreeCoordinate::from_addrs(ids.iter().map(|&v| make_node_addr(v)).collect()).unwrap()
1167 }
1168
1169 #[test]
1172 fn test_session_message_type_roundtrip() {
1173 let types = [
1174 SessionMessageType::DataPacket,
1175 SessionMessageType::SenderReport,
1176 SessionMessageType::ReceiverReport,
1177 SessionMessageType::PathMtuNotification,
1178 SessionMessageType::CoordsWarmup,
1179 SessionMessageType::EndpointData,
1180 SessionMessageType::CoordsRequired,
1181 SessionMessageType::PathBroken,
1182 SessionMessageType::MtuExceeded,
1183 ];
1184
1185 for ty in types {
1186 let byte = ty.to_byte();
1187 let restored = SessionMessageType::from_byte(byte);
1188 assert_eq!(restored, Some(ty));
1189 }
1190 }
1191
1192 #[test]
1193 fn test_session_message_type_invalid() {
1194 assert!(SessionMessageType::from_byte(0xFF).is_none());
1195 assert!(SessionMessageType::from_byte(0x99).is_none());
1196 }
1197
1198 #[test]
1201 fn test_session_flags() {
1202 let flags = SessionFlags::new().with_ack().bidirectional();
1203
1204 assert!(flags.request_ack);
1205 assert!(flags.bidirectional);
1206
1207 let byte = flags.to_byte();
1208 let restored = SessionFlags::from_byte(byte);
1209
1210 assert_eq!(flags, restored);
1211 }
1212
1213 #[test]
1214 fn test_session_flags_default() {
1215 let flags = SessionFlags::new();
1216 assert!(!flags.request_ack);
1217 assert!(!flags.bidirectional);
1218 assert_eq!(flags.to_byte(), 0);
1219 }
1220
1221 #[test]
1224 fn test_session_setup() {
1225 let setup = SessionSetup::new(make_coords(&[1, 0]), make_coords(&[2, 0]))
1226 .with_flags(SessionFlags::new().with_ack());
1227
1228 assert!(setup.flags.request_ack);
1229 assert!(!setup.flags.bidirectional);
1230 }
1231
1232 #[test]
1235 fn test_coords_required() {
1236 let err = CoordsRequired::new(make_node_addr(1), make_node_addr(2));
1237
1238 assert_eq!(err.dest_addr, make_node_addr(1));
1239 assert_eq!(err.reporter, make_node_addr(2));
1240 }
1241
1242 #[test]
1245 fn test_path_broken() {
1246 let err = PathBroken::new(make_node_addr(2), make_node_addr(3))
1247 .with_last_coords(make_coords(&[2, 0]));
1248
1249 assert_eq!(err.dest_addr, make_node_addr(2));
1250 assert_eq!(err.reporter, make_node_addr(3));
1251 assert!(err.last_known_coords.is_some());
1252 }
1253
1254 #[test]
1257 fn test_session_setup_encode_decode() {
1258 let handshake = vec![0xAA; 82]; let setup = SessionSetup::new(make_coords(&[1, 2, 0]), make_coords(&[3, 4, 0]))
1260 .with_flags(SessionFlags::new().with_ack().bidirectional())
1261 .with_handshake(handshake.clone());
1262
1263 let encoded = setup.encode();
1264
1265 assert_eq!(encoded[0], 0x01);
1267 assert_eq!(encoded[1], 0x00); let payload_len = u16::from_le_bytes([encoded[2], encoded[3]]);
1269 assert_eq!(payload_len as usize, encoded.len() - 4);
1270
1271 let decoded = SessionSetup::decode(&encoded[4..]).unwrap();
1273
1274 assert_eq!(decoded.flags, setup.flags);
1275 assert_eq!(decoded.src_coords, setup.src_coords);
1276 assert_eq!(decoded.dest_coords, setup.dest_coords);
1277 assert_eq!(decoded.handshake_payload, handshake);
1278 }
1279
1280 #[test]
1281 fn test_session_setup_no_handshake() {
1282 let setup = SessionSetup::new(make_coords(&[5, 0]), make_coords(&[6, 0]));
1283
1284 let encoded = setup.encode();
1285 let decoded = SessionSetup::decode(&encoded[4..]).unwrap();
1286
1287 assert!(decoded.handshake_payload.is_empty());
1288 assert_eq!(decoded.src_coords, setup.src_coords);
1289 assert_eq!(decoded.dest_coords, setup.dest_coords);
1290 }
1291
1292 #[test]
1293 fn test_session_ack_encode_decode() {
1294 let handshake = vec![0xBB; 33]; let ack = SessionAck::new(make_coords(&[7, 8, 0]), make_coords(&[3, 4, 0]))
1296 .with_handshake(handshake.clone());
1297
1298 let encoded = ack.encode();
1299 assert_eq!(encoded[0], 0x02);
1301 assert_eq!(encoded[1], 0x00); let decoded = SessionAck::decode(&encoded[4..]).unwrap();
1304 assert_eq!(decoded.src_coords, ack.src_coords);
1305 assert_eq!(decoded.dest_coords, ack.dest_coords);
1306 assert_eq!(decoded.handshake_payload, handshake);
1307 }
1308
1309 #[test]
1310 fn test_coords_required_encode_decode() {
1311 let err = CoordsRequired::new(make_node_addr(0xAA), make_node_addr(0xBB));
1312
1313 let encoded = err.encode();
1314 assert_eq!(encoded.len(), 4 + COORDS_REQUIRED_SIZE);
1316 assert_eq!(encoded[0], 0x00);
1318 assert_eq!(encoded[1], 0x04); assert_eq!(encoded[4], 0x20);
1321
1322 let decoded = CoordsRequired::decode(&encoded[5..]).unwrap();
1324 assert_eq!(decoded.dest_addr, err.dest_addr);
1325 assert_eq!(decoded.reporter, err.reporter);
1326 }
1327
1328 #[test]
1329 fn test_path_broken_encode_decode_no_coords() {
1330 let err = PathBroken::new(make_node_addr(0xCC), make_node_addr(0xDD));
1331
1332 let encoded = err.encode();
1333 assert_eq!(encoded[0], 0x00);
1335 assert_eq!(encoded[1], 0x04); assert_eq!(encoded[4], 0x21); let decoded = PathBroken::decode(&encoded[5..]).unwrap();
1339 assert_eq!(decoded.dest_addr, err.dest_addr);
1340 assert_eq!(decoded.reporter, err.reporter);
1341 assert!(decoded.last_known_coords.is_none());
1342 }
1343
1344 #[test]
1345 fn test_path_broken_encode_decode_with_coords() {
1346 let coords = make_coords(&[0xCC, 0xDD, 0xEE]);
1347 let err = PathBroken::new(make_node_addr(0x11), make_node_addr(0x22))
1348 .with_last_coords(coords.clone());
1349
1350 let encoded = err.encode();
1351 let decoded = PathBroken::decode(&encoded[5..]).unwrap();
1352
1353 assert_eq!(decoded.dest_addr, err.dest_addr);
1354 assert_eq!(decoded.reporter, err.reporter);
1355 assert_eq!(decoded.last_known_coords.unwrap(), coords);
1356 }
1357
1358 #[test]
1359 fn test_session_setup_decode_too_short() {
1360 assert!(SessionSetup::decode(&[]).is_err());
1361 }
1362
1363 #[test]
1364 fn test_session_ack_decode_too_short() {
1365 assert!(SessionAck::decode(&[]).is_err());
1366 }
1367
1368 #[test]
1369 fn test_coords_required_decode_too_short() {
1370 assert!(CoordsRequired::decode(&[]).is_err());
1371 assert!(CoordsRequired::decode(&[0x00; 10]).is_err());
1372 }
1373
1374 #[test]
1375 fn test_path_broken_decode_too_short() {
1376 assert!(PathBroken::decode(&[]).is_err());
1377 assert!(PathBroken::decode(&[0x00; 20]).is_err());
1378 }
1379
1380 #[test]
1381 fn test_session_setup_deep_coords() {
1382 let addrs: Vec<u8> = (0..11).collect();
1384 let src = make_coords(&addrs);
1385 let dest = make_coords(&[20, 21, 22, 23, 24]);
1386 let setup = SessionSetup::new(src.clone(), dest.clone()).with_handshake(vec![0x55; 82]);
1387
1388 let encoded = setup.encode();
1389 let decoded = SessionSetup::decode(&encoded[4..]).unwrap();
1390
1391 assert_eq!(decoded.src_coords, src);
1392 assert_eq!(decoded.dest_coords, dest);
1393 }
1394
1395 #[test]
1398 fn test_fsp_flags_default() {
1399 let flags = FspFlags::new();
1400 assert!(!flags.coords_present);
1401 assert!(!flags.key_epoch);
1402 assert!(!flags.unencrypted);
1403 assert_eq!(flags.to_byte(), 0x00);
1404 }
1405
1406 #[test]
1407 fn test_fsp_flags_roundtrip() {
1408 for byte in 0u8..=0x07 {
1410 let flags = FspFlags::from_byte(byte);
1411 assert_eq!(flags.to_byte(), byte);
1412 }
1413 }
1414
1415 #[test]
1416 fn test_fsp_flags_individual_bits() {
1417 let cp = FspFlags::from_byte(0x01);
1418 assert!(cp.coords_present);
1419 assert!(!cp.key_epoch);
1420 assert!(!cp.unencrypted);
1421
1422 let k = FspFlags::from_byte(0x02);
1423 assert!(!k.coords_present);
1424 assert!(k.key_epoch);
1425 assert!(!k.unencrypted);
1426
1427 let u = FspFlags::from_byte(0x04);
1428 assert!(!u.coords_present);
1429 assert!(!u.key_epoch);
1430 assert!(u.unencrypted);
1431 }
1432
1433 #[test]
1434 fn test_fsp_flags_ignores_reserved_bits() {
1435 let flags = FspFlags::from_byte(0xFF);
1437 assert!(flags.coords_present);
1438 assert!(flags.key_epoch);
1439 assert!(flags.unencrypted);
1440 assert_eq!(flags.to_byte(), 0x07); }
1442
1443 #[test]
1446 fn test_fsp_inner_flags_default() {
1447 let flags = FspInnerFlags::new();
1448 assert!(!flags.spin_bit);
1449 assert_eq!(flags.to_byte(), 0x00);
1450 }
1451
1452 #[test]
1453 fn test_fsp_inner_flags_roundtrip() {
1454 let flags = FspInnerFlags::from_byte(0x01);
1455 assert!(flags.spin_bit);
1456 assert_eq!(flags.to_byte(), 0x01);
1457
1458 let flags = FspInnerFlags::from_byte(0x00);
1459 assert!(!flags.spin_bit);
1460 assert_eq!(flags.to_byte(), 0x00);
1461 }
1462
1463 #[test]
1464 fn test_fsp_inner_flags_ignores_reserved() {
1465 let flags = FspInnerFlags::from_byte(0xFE);
1466 assert!(!flags.spin_bit);
1467 assert_eq!(flags.to_byte(), 0x00);
1468
1469 let flags = FspInnerFlags::from_byte(0xFF);
1470 assert!(flags.spin_bit);
1471 assert_eq!(flags.to_byte(), 0x01);
1472 }
1473
1474 #[test]
1477 fn test_session_message_type_new_values() {
1478 assert_eq!(SessionMessageType::SenderReport.to_byte(), 0x11);
1479 assert_eq!(SessionMessageType::ReceiverReport.to_byte(), 0x12);
1480 assert_eq!(SessionMessageType::PathMtuNotification.to_byte(), 0x13);
1481 assert_eq!(SessionMessageType::CoordsWarmup.to_byte(), 0x14);
1482 assert_eq!(SessionMessageType::EndpointData.to_byte(), 0x15);
1483 }
1484
1485 #[test]
1486 fn test_session_message_type_display() {
1487 assert_eq!(
1488 format!("{}", SessionMessageType::SenderReport),
1489 "SenderReport"
1490 );
1491 assert_eq!(
1492 format!("{}", SessionMessageType::ReceiverReport),
1493 "ReceiverReport"
1494 );
1495 assert_eq!(
1496 format!("{}", SessionMessageType::PathMtuNotification),
1497 "PathMtuNotification"
1498 );
1499 assert_eq!(
1500 format!("{}", SessionMessageType::CoordsWarmup),
1501 "CoordsWarmup"
1502 );
1503 assert_eq!(
1504 format!("{}", SessionMessageType::EndpointData),
1505 "EndpointData"
1506 );
1507 }
1508
1509 fn sample_session_sender_report() -> SessionSenderReport {
1512 SessionSenderReport {
1513 interval_start_counter: 100,
1514 interval_end_counter: 200,
1515 interval_start_timestamp: 5000,
1516 interval_end_timestamp: 6000,
1517 interval_bytes_sent: 50_000,
1518 cumulative_packets_sent: 10_000,
1519 cumulative_bytes_sent: 5_000_000,
1520 }
1521 }
1522
1523 #[test]
1524 fn test_session_sender_report_encode_size() {
1525 let sr = sample_session_sender_report();
1526 let encoded = sr.encode();
1527 assert_eq!(encoded.len(), SESSION_SENDER_REPORT_SIZE);
1528 }
1529
1530 #[test]
1531 fn test_session_sender_report_roundtrip() {
1532 let sr = sample_session_sender_report();
1533 let encoded = sr.encode();
1534 let decoded = SessionSenderReport::decode(&encoded).unwrap();
1535 assert_eq!(sr, decoded);
1536 }
1537
1538 #[test]
1539 fn test_session_sender_report_too_short() {
1540 assert!(SessionSenderReport::decode(&[0u8; 10]).is_err());
1541 }
1542
1543 fn sample_session_receiver_report() -> SessionReceiverReport {
1546 SessionReceiverReport {
1547 highest_counter: 195,
1548 cumulative_packets_recv: 9_500,
1549 cumulative_bytes_recv: 4_750_000,
1550 timestamp_echo: 5900,
1551 dwell_time: 5,
1552 max_burst_loss: 3,
1553 mean_burst_loss: 384,
1554 jitter: 1200,
1555 ecn_ce_count: 0,
1556 owd_trend: -50,
1557 burst_loss_count: 2,
1558 cumulative_reorder_count: 10,
1559 interval_packets_recv: 95,
1560 interval_bytes_recv: 47_500,
1561 }
1562 }
1563
1564 #[test]
1565 fn test_session_receiver_report_encode_size() {
1566 let rr = sample_session_receiver_report();
1567 let encoded = rr.encode();
1568 assert_eq!(encoded.len(), SESSION_RECEIVER_REPORT_SIZE);
1569 }
1570
1571 #[test]
1572 fn test_session_receiver_report_roundtrip() {
1573 let rr = sample_session_receiver_report();
1574 let encoded = rr.encode();
1575 let decoded = SessionReceiverReport::decode(&encoded).unwrap();
1576 assert_eq!(rr, decoded);
1577 }
1578
1579 #[test]
1580 fn test_session_receiver_report_too_short() {
1581 assert!(SessionReceiverReport::decode(&[0u8; 10]).is_err());
1582 }
1583
1584 #[test]
1585 fn test_session_receiver_report_negative_owd_trend() {
1586 let rr = SessionReceiverReport {
1587 owd_trend: -12345,
1588 ..sample_session_receiver_report()
1589 };
1590 let encoded = rr.encode();
1591 let decoded = SessionReceiverReport::decode(&encoded).unwrap();
1592 assert_eq!(decoded.owd_trend, -12345);
1593 }
1594
1595 #[test]
1598 fn test_path_mtu_notification_encode_size() {
1599 let n = PathMtuNotification::new(1400);
1600 let encoded = n.encode();
1601 assert_eq!(encoded.len(), PATH_MTU_NOTIFICATION_SIZE);
1602 }
1603
1604 #[test]
1605 fn test_path_mtu_notification_roundtrip() {
1606 let n = PathMtuNotification::new(1400);
1607 let encoded = n.encode();
1608 let decoded = PathMtuNotification::decode(&encoded).unwrap();
1609 assert_eq!(decoded.path_mtu, 1400);
1610 }
1611
1612 #[test]
1613 fn test_path_mtu_notification_too_short() {
1614 assert!(PathMtuNotification::decode(&[]).is_err());
1615 assert!(PathMtuNotification::decode(&[0x00]).is_err());
1616 }
1617
1618 #[test]
1619 fn test_path_mtu_notification_boundary_values() {
1620 for mtu in [0u16, 1280, 1500, u16::MAX] {
1621 let n = PathMtuNotification::new(mtu);
1622 let encoded = n.encode();
1623 let decoded = PathMtuNotification::decode(&encoded).unwrap();
1624 assert_eq!(decoded.path_mtu, mtu);
1625 }
1626 }
1627
1628 #[test]
1631 fn test_mtu_exceeded_encode_size() {
1632 let err = MtuExceeded::new(make_node_addr(0xAA), make_node_addr(0xBB), 1400);
1633 let encoded = err.encode();
1634 assert_eq!(encoded.len(), 4 + MTU_EXCEEDED_SIZE);
1636 }
1637
1638 #[test]
1639 fn test_mtu_exceeded_encode_decode() {
1640 let err = MtuExceeded::new(make_node_addr(0xAA), make_node_addr(0xBB), 1400);
1641
1642 let encoded = err.encode();
1643 assert_eq!(encoded[0], 0x00);
1645 assert_eq!(encoded[1], 0x04); assert_eq!(encoded[4], 0x22);
1648
1649 let decoded = MtuExceeded::decode(&encoded[5..]).unwrap();
1651 assert_eq!(decoded.dest_addr, err.dest_addr);
1652 assert_eq!(decoded.reporter, err.reporter);
1653 assert_eq!(decoded.mtu, 1400);
1654 }
1655
1656 #[test]
1657 fn test_mtu_exceeded_decode_too_short() {
1658 assert!(MtuExceeded::decode(&[]).is_err());
1659 assert!(MtuExceeded::decode(&[0x00; 20]).is_err());
1660 assert!(MtuExceeded::decode(&[0x00; 34]).is_err()); }
1662
1663 #[test]
1664 fn test_mtu_exceeded_boundary_mtu_values() {
1665 for mtu in [0u16, 1280, 1500, u16::MAX] {
1666 let err = MtuExceeded::new(make_node_addr(1), make_node_addr(2), mtu);
1667 let encoded = err.encode();
1668 let decoded = MtuExceeded::decode(&encoded[5..]).unwrap();
1669 assert_eq!(decoded.mtu, mtu);
1670 }
1671 }
1672
1673 #[test]
1674 fn test_mtu_exceeded_message_type_value() {
1675 assert_eq!(SessionMessageType::MtuExceeded.to_byte(), 0x22);
1676 assert_eq!(
1677 SessionMessageType::from_byte(0x22),
1678 Some(SessionMessageType::MtuExceeded)
1679 );
1680 }
1681
1682 #[test]
1683 fn test_mtu_exceeded_display() {
1684 assert_eq!(
1685 format!("{}", SessionMessageType::MtuExceeded),
1686 "MtuExceeded"
1687 );
1688 }
1689
1690 #[test]
1693 fn test_session_msg3_encode_decode() {
1694 let handshake = vec![0xCC; 73]; let msg3 = SessionMsg3::new(handshake.clone());
1696
1697 let encoded = msg3.encode();
1698 assert_eq!(encoded[0], 0x03);
1700 assert_eq!(encoded[1], 0x00); let payload_len = u16::from_le_bytes([encoded[2], encoded[3]]);
1702 assert_eq!(payload_len as usize, encoded.len() - 4);
1703
1704 let decoded = SessionMsg3::decode(&encoded[4..]).unwrap();
1706 assert_eq!(decoded.flags, 0);
1707 assert_eq!(decoded.handshake_payload, handshake);
1708 }
1709
1710 #[test]
1711 fn test_session_msg3_decode_too_short() {
1712 assert!(SessionMsg3::decode(&[]).is_err());
1713 assert!(SessionMsg3::decode(&[0x00]).is_err()); }
1715
1716 #[test]
1717 fn test_session_msg3_empty_handshake() {
1718 let msg3 = SessionMsg3::new(vec![]);
1719 let encoded = msg3.encode();
1720 let decoded = SessionMsg3::decode(&encoded[4..]).unwrap();
1721 assert!(decoded.handshake_payload.is_empty());
1722 }
1723}