diff --git a/src/control/queries.rs b/src/control/queries.rs index 0ec2475..6325d32 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -373,7 +373,6 @@ pub fn show_mmp(node: &Node) -> Value { "loss_rate": metrics.loss_rate(), "etx": metrics.etx, "goodput_bps": metrics.goodput_bps, - "spin_bit_role": if mmp.spin_bit.is_initiator() { "initiator" } else { "responder" }, }); if let Some(smoothed_loss) = metrics.smoothed_loss() { diff --git a/src/mmp/algorithms.rs b/src/mmp/algorithms.rs index 080af10..cc11d86 100644 --- a/src/mmp/algorithms.rs +++ b/src/mmp/algorithms.rs @@ -4,7 +4,6 @@ //! Each is independently testable. use std::collections::VecDeque; -use std::time::Instant; use crate::mmp::{EWMA_LONG_ALPHA, EWMA_SHORT_ALPHA}; @@ -278,75 +277,6 @@ pub fn compute_etx(d_forward: f64, d_reverse: f64) -> f64 { // Spin Bit // ============================================================================ -/// Spin bit state for passive RTT estimation. -/// -/// Uses asymmetric roles (initiator/responder) per the MMP design: -/// - **Initiator**: flips spin value on each received frame; measures RTT -/// from edge-to-edge intervals. -/// - **Responder**: copies received spin bit into outgoing frames, with a -/// counter guard to filter reordered frames. -pub struct SpinBitState { - is_initiator: bool, - current_value: bool, - /// Highest counter observed with a spin edge (responder guard). - highest_counter_for_spin: u64, - /// Time of last spin edge (initiator only, for RTT measurement). - last_edge_time: Option, -} - -impl SpinBitState { - pub fn new(is_initiator: bool) -> Self { - Self { - is_initiator, - current_value: false, - highest_counter_for_spin: 0, - last_edge_time: None, - } - } - - /// Check if this is the spin bit initiator. - pub fn is_initiator(&self) -> bool { - self.is_initiator - } - - /// Get the spin bit value to set on an outgoing frame. - pub fn tx_bit(&self) -> bool { - self.current_value - } - - /// Process a received frame's spin bit. - /// - /// Returns an RTT sample duration if an edge was detected (initiator only). - pub fn rx_observe( - &mut self, - received_bit: bool, - counter: u64, - now: Instant, - ) -> Option { - if self.is_initiator { - // Initiator: when the reflected bit matches what we sent, - // that completes a round trip. Record the edge time, then - // flip for the next cycle. - if received_bit == self.current_value { - let rtt = self.last_edge_time.map(|t| now.duration_since(t)); - self.last_edge_time = Some(now); - self.current_value = !self.current_value; - rtt - } else { - None - } - } else { - // Responder: copy received bit, but only if counter is higher - // (reordering guard) - if counter > self.highest_counter_for_spin { - self.highest_counter_for_spin = counter; - self.current_value = received_bit; - } - None - } - } -} - // ============================================================================ // Tests // ============================================================================ @@ -473,54 +403,4 @@ mod tests { assert_eq!(compute_etx(1.0, 0.0), 100.0); } - #[test] - fn test_spin_bit_initiator_rtt() { - let mut initiator = SpinBitState::new(true); - let mut responder = SpinBitState::new(false); - - let t0 = Instant::now(); - let t1 = t0 + std::time::Duration::from_millis(10); - let t2 = t0 + std::time::Duration::from_millis(20); - - // Initiator sends with spin=false (initial) - let bit_to_send = initiator.tx_bit(); - assert!(!bit_to_send); - - // Responder receives, copies bit - responder.rx_observe(bit_to_send, 1, t0); - assert!(!responder.tx_bit()); - - // Responder sends back, initiator receives - let resp_bit = responder.tx_bit(); - let rtt1 = initiator.rx_observe(resp_bit, 2, t1); - // First edge: no previous edge to compare - assert!(rtt1.is_none()); - - // Now initiator's spin flipped to true - let bit2 = initiator.tx_bit(); - assert!(bit2); - - // Responder receives new bit - responder.rx_observe(bit2, 3, t1); - assert!(responder.tx_bit()); - - // Responder sends back, initiator receives - let resp_bit2 = responder.tx_bit(); - let rtt2 = initiator.rx_observe(resp_bit2, 4, t2); - // Second edge: should produce an RTT sample - assert!(rtt2.is_some()); - } - - #[test] - fn test_spin_bit_responder_counter_guard() { - let mut responder = SpinBitState::new(false); - - // Receive counter=5 with spin=true - responder.rx_observe(true, 5, Instant::now()); - assert!(responder.tx_bit()); - - // Reordered packet with counter=3 and spin=false should be ignored - responder.rx_observe(false, 3, Instant::now()); - assert!(responder.tx_bit()); // unchanged - } } diff --git a/src/mmp/metrics.rs b/src/mmp/metrics.rs index c12fef7..e6b10a3 100644 --- a/src/mmp/metrics.rs +++ b/src/mmp/metrics.rs @@ -287,20 +287,16 @@ mod tests { jitter: u32, ) -> ReceiverReport { ReceiverReport { + timestamp_echo, + dwell_time: dwell, highest_counter, cumulative_packets_recv: cum_packets, cumulative_bytes_recv: cum_bytes, - timestamp_echo, - dwell_time: dwell, - max_burst_loss: 0, - mean_burst_loss: 0, jitter, ecn_ce_count: 0, owd_trend: 0, burst_loss_count: 0, cumulative_reorder_count: 0, - interval_packets_recv: 0, - interval_bytes_recv: 0, } } diff --git a/src/mmp/mod.rs b/src/mmp/mod.rs index b894062..2d9beac 100644 --- a/src/mmp/mod.rs +++ b/src/mmp/mod.rs @@ -23,7 +23,7 @@ pub mod sender; // Re-exports pub use algorithms::{ - DualEwma, JitterEstimator, OwdTrendDetector, SpinBitState, SrttEstimator, compute_etx, + DualEwma, JitterEstimator, OwdTrendDetector, SrttEstimator, compute_etx, }; pub use metrics::MmpMetrics; pub use receiver::ReceiverState; @@ -180,13 +180,12 @@ impl MmpConfig { /// Combined MMP state for a single peer link. /// -/// Wraps sender, receiver, metrics, and spin bit state. One instance +/// Wraps sender, receiver, and metrics state. One instance /// per `ActivePeer`. pub struct MmpPeerState { pub sender: SenderState, pub receiver: ReceiverState, pub metrics: MmpMetrics, - pub spin_bit: SpinBitState, mode: MmpMode, log_interval: Duration, last_log_time: Option, @@ -194,15 +193,12 @@ pub struct MmpPeerState { impl MmpPeerState { /// Create MMP state for a new peer link. - /// - /// `is_initiator`: true if this node initiated the Noise handshake - /// (determines spin bit role). pub fn new(config: &MmpConfig, is_initiator: bool) -> Self { + let _ = is_initiator; // retained for API compatibility Self { sender: SenderState::new(), receiver: ReceiverState::new(config.owd_window_size), metrics: MmpMetrics::new(), - spin_bit: SpinBitState::new(is_initiator), mode: config.mode, log_interval: Duration::from_secs(config.log_interval_secs), last_log_time: None, @@ -240,13 +236,12 @@ impl MmpPeerState { /// Combined MMP state for a single end-to-end session. /// -/// Wraps sender, receiver, metrics, spin bit, and path MTU state. +/// Wraps sender, receiver, metrics, and path MTU state. /// One instance per established `SessionEntry`. pub struct MmpSessionState { pub sender: SenderState, pub receiver: ReceiverState, pub metrics: MmpMetrics, - pub spin_bit: SpinBitState, mode: MmpMode, log_interval: Duration, last_log_time: Option, @@ -255,10 +250,8 @@ pub struct MmpSessionState { impl MmpSessionState { /// Create MMP state for a new session. - /// - /// `is_initiator`: true if this node initiated the Noise handshake - /// (determines spin bit role). pub fn new(config: &crate::config::SessionMmpConfig, is_initiator: bool) -> Self { + let _ = is_initiator; // retained for API compatibility Self { sender: SenderState::new_with_cold_start(SESSION_COLD_START_INTERVAL_MS), receiver: ReceiverState::new_with_cold_start( @@ -266,7 +259,6 @@ impl MmpSessionState { SESSION_COLD_START_INTERVAL_MS, ), metrics: MmpMetrics::new(), - spin_bit: SpinBitState::new(is_initiator), mode: config.mode, log_interval: Duration::from_secs(config.log_interval_secs), last_log_time: None, diff --git a/src/mmp/receiver.rs b/src/mmp/receiver.rs index 0582431..0a6fc47 100644 --- a/src/mmp/receiver.rs +++ b/src/mmp/receiver.rs @@ -329,23 +329,19 @@ impl ReceiverState { .map(|t| now.duration_since(t).as_millis() as u16) .unwrap_or(0); - let (burst_count, max_burst, mean_burst) = self.gap_tracker.take_interval_stats(); + let (burst_count, _max_burst, _mean_burst) = self.gap_tracker.take_interval_stats(); let report = ReceiverReport { + timestamp_echo: self.last_sender_timestamp, + dwell_time, highest_counter: self.highest_counter, cumulative_packets_recv: self.cumulative_packets_recv, cumulative_bytes_recv: self.cumulative_bytes_recv, - timestamp_echo: self.last_sender_timestamp, - dwell_time, - max_burst_loss: max_burst, - mean_burst_loss: mean_burst, jitter: self.jitter.jitter_us(), ecn_ce_count: self.ecn_ce_count, owd_trend: self.owd_trend.trend_us_per_sec(), burst_loss_count: burst_count, cumulative_reorder_count: self.cumulative_reorder_count as u32, - interval_packets_recv: self.interval_packets_recv, - interval_bytes_recv: self.interval_bytes_recv, }; // Reset interval @@ -501,8 +497,6 @@ mod tests { assert_eq!(report.cumulative_packets_recv, 2); assert_eq!(report.cumulative_bytes_recv, 1100); assert_eq!(report.timestamp_echo, 200); // last sender timestamp - assert_eq!(report.interval_packets_recv, 2); - assert_eq!(report.interval_bytes_recv, 1100); } #[test] @@ -518,8 +512,6 @@ mod tests { // New data r.record_recv(2, 200, 300, false, t0 + Duration::from_millis(100)); let report = r.build_report(t0 + Duration::from_millis(150)).unwrap(); - assert_eq!(report.interval_packets_recv, 1); - assert_eq!(report.interval_bytes_recv, 300); // Cumulative continues assert_eq!(report.cumulative_packets_recv, 2); } diff --git a/src/mmp/report.rs b/src/mmp/report.rs index 8d0289f..2ba0cda 100644 --- a/src/mmp/report.rs +++ b/src/mmp/report.rs @@ -1,173 +1,203 @@ //! MMP report wire format: SenderReport and ReceiverReport. //! //! Serialization and deserialization for the two report types exchanged -//! between link-layer peers. Wire format follows the MMP design doc. +//! between link-layer peers. Wire format uses an extensibility header: +//! `[msg_type:1][format_version:1][total_length:2 LE]` followed by payload. +//! Format version 0 defines the slim layouts below. Decoders skip unknown +//! trailing bytes via total_length for forward compatibility. use crate::protocol::ProtocolError; +/// Current format version for MMP reports. +const FORMAT_VERSION: u8 = 0; + // ============================================================================ -// SenderReport (msg_type 0x01, 48-byte body including type byte) +// SenderReport (msg_type 0x01, 20 bytes total) // ============================================================================ /// Link-layer sender report. /// -/// Wire layout (48 bytes total, sent as link message): +/// Wire layout (20 bytes total, sent as link message): /// ```text -/// [0] msg_type = 0x01 -/// [1-3] reserved (zero) -/// [4-11] interval_start_counter: u64 LE -/// [12-19] interval_end_counter: u64 LE -/// [20-23] interval_start_timestamp: u32 LE -/// [24-27] interval_end_timestamp: u32 LE -/// [28-31] interval_bytes_sent: u32 LE -/// [32-39] cumulative_packets_sent: u64 LE -/// [40-47] cumulative_bytes_sent: u64 LE +/// [0] msg_type = 0x01 +/// [1] format_version = 0 +/// [2-3] total_length: u16 LE (= 16, payload bytes after this field) +/// [4-7] interval_packets_sent: u32 LE +/// [8-11] interval_bytes_sent: u32 LE +/// [12-19] cumulative_packets_sent: u64 LE /// ``` #[derive(Debug, Clone, PartialEq, Eq)] pub struct SenderReport { - pub interval_start_counter: u64, - pub interval_end_counter: u64, - pub interval_start_timestamp: u32, - pub interval_end_timestamp: u32, + pub interval_packets_sent: u32, pub interval_bytes_sent: u32, pub cumulative_packets_sent: u64, - pub cumulative_bytes_sent: u64, } -/// ReceiverReport (msg_type 0x02, 68-byte body including type byte) +/// Total wire size for SenderReport. +pub const SENDER_REPORT_SIZE: usize = 20; + +/// Payload size after total_length field for SenderReport format v0. +const SENDER_REPORT_PAYLOAD: u16 = 16; + +/// ReceiverReport (msg_type 0x02, 54 bytes total) /// -/// Wire layout (68 bytes total, sent as link message): +/// Wire layout (54 bytes total, sent as link message): /// ```text -/// [0] msg_type = 0x02 -/// [1-3] reserved (zero) -/// [4-11] highest_counter: u64 LE -/// [12-19] cumulative_packets_recv: u64 LE -/// [20-27] cumulative_bytes_recv: u64 LE -/// [28-31] timestamp_echo: u32 LE -/// [32-33] dwell_time: u16 LE -/// [34-35] max_burst_loss: u16 LE -/// [36-37] mean_burst_loss: u16 LE (u8.8 fixed-point) -/// [38-39] reserved: u16 LE -/// [40-43] jitter: u32 LE (microseconds) -/// [44-47] ecn_ce_count: u32 LE -/// [48-51] owd_trend: i32 LE (µs/s) -/// [52-55] burst_loss_count: u32 LE -/// [56-59] cumulative_reorder_count: u32 LE -/// [60-63] interval_packets_recv: u32 LE -/// [64-67] interval_bytes_recv: u32 LE +/// [0] msg_type = 0x02 +/// [1] format_version = 0 +/// [2-3] total_length: u16 LE (= 50, payload bytes after this field) +/// [4-7] timestamp_echo: u32 LE +/// [8-9] dwell_time: u16 LE +/// [10-17] highest_counter: u64 LE +/// [18-25] cumulative_packets_recv: u64 LE +/// [26-33] cumulative_bytes_recv: u64 LE +/// [34-37] jitter: u32 LE (microseconds) +/// [38-41] ecn_ce_count: u32 LE +/// [42-45] owd_trend: i32 LE (µs/s) +/// [46-49] burst_loss_count: u32 LE +/// [50-53] cumulative_reorder_count: u32 LE /// ``` #[derive(Debug, Clone, PartialEq, Eq)] pub struct ReceiverReport { + pub timestamp_echo: u32, + pub dwell_time: u16, pub highest_counter: u64, pub cumulative_packets_recv: u64, pub cumulative_bytes_recv: u64, - pub timestamp_echo: u32, - pub dwell_time: u16, - pub max_burst_loss: u16, - pub mean_burst_loss: u16, pub jitter: u32, pub ecn_ce_count: u32, pub owd_trend: i32, pub burst_loss_count: u32, pub cumulative_reorder_count: u32, - pub interval_packets_recv: u32, - pub interval_bytes_recv: u32, } -// Encode/decode will be implemented in Step 2. +/// Total wire size for ReceiverReport. +pub const RECEIVER_REPORT_SIZE: usize = 54; + +/// Payload size after total_length field for ReceiverReport format v0. +const RECEIVER_REPORT_PAYLOAD: u16 = 50; impl SenderReport { - /// Encode to wire format (48 bytes: msg_type + 3 reserved + 44 payload). + /// Encode to wire format (20 bytes: header + payload). pub fn encode(&self) -> Vec { - let mut buf = Vec::with_capacity(48); + let mut buf = Vec::with_capacity(SENDER_REPORT_SIZE); buf.push(0x01); // msg_type - buf.extend_from_slice(&[0u8; 3]); // reserved - buf.extend_from_slice(&self.interval_start_counter.to_le_bytes()); - buf.extend_from_slice(&self.interval_end_counter.to_le_bytes()); - buf.extend_from_slice(&self.interval_start_timestamp.to_le_bytes()); - buf.extend_from_slice(&self.interval_end_timestamp.to_le_bytes()); + buf.push(FORMAT_VERSION); + buf.extend_from_slice(&SENDER_REPORT_PAYLOAD.to_le_bytes()); + buf.extend_from_slice(&self.interval_packets_sent.to_le_bytes()); buf.extend_from_slice(&self.interval_bytes_sent.to_le_bytes()); buf.extend_from_slice(&self.cumulative_packets_sent.to_le_bytes()); - buf.extend_from_slice(&self.cumulative_bytes_sent.to_le_bytes()); buf } /// Decode from payload after msg_type byte has been consumed. /// - /// `payload` starts at the reserved bytes (offset 1 in the wire format). + /// `payload` starts at format_version (offset 1 in the wire format). + /// Unknown trailing bytes (from future format extensions) are skipped + /// via total_length. pub fn decode(payload: &[u8]) -> Result { - if payload.len() < 47 { + // Need at least: format_version(1) + total_length(2) + v0 payload(16) = 19 + if payload.len() < 19 { return Err(ProtocolError::MessageTooShort { - expected: 47, + expected: 19, got: payload.len(), }); } - // Skip 3 reserved bytes + let format_version = payload[0]; + let total_length = u16::from_le_bytes(payload[1..3].try_into().unwrap()) as usize; + + // Verify we have enough data for the declared length + if payload.len() < 3 + total_length { + return Err(ProtocolError::MessageTooShort { + expected: 3 + total_length, + got: payload.len(), + }); + } + + // For version 0, parse known fields from offset 3 + if format_version > 0 { + // Future versions: we can still parse v0 fields if total_length >= 14 + if total_length < SENDER_REPORT_PAYLOAD as usize { + return Err(ProtocolError::MessageTooShort { + expected: SENDER_REPORT_PAYLOAD as usize, + got: total_length, + }); + } + } + let p = &payload[3..]; Ok(Self { - interval_start_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()), - interval_end_counter: u64::from_le_bytes(p[8..16].try_into().unwrap()), - interval_start_timestamp: u32::from_le_bytes(p[16..20].try_into().unwrap()), - interval_end_timestamp: u32::from_le_bytes(p[20..24].try_into().unwrap()), - interval_bytes_sent: u32::from_le_bytes(p[24..28].try_into().unwrap()), - cumulative_packets_sent: u64::from_le_bytes(p[28..36].try_into().unwrap()), - cumulative_bytes_sent: u64::from_le_bytes(p[36..44].try_into().unwrap()), + interval_packets_sent: u32::from_le_bytes(p[0..4].try_into().unwrap()), + interval_bytes_sent: u32::from_le_bytes(p[4..8].try_into().unwrap()), + cumulative_packets_sent: u64::from_le_bytes(p[8..16].try_into().unwrap()), }) } } impl ReceiverReport { - /// Encode to wire format (68 bytes: msg_type + 3 reserved + 64 payload). + /// Encode to wire format (54 bytes: header + payload). pub fn encode(&self) -> Vec { - let mut buf = Vec::with_capacity(68); + let mut buf = Vec::with_capacity(RECEIVER_REPORT_SIZE); buf.push(0x02); // msg_type - buf.extend_from_slice(&[0u8; 3]); // reserved + buf.push(FORMAT_VERSION); + buf.extend_from_slice(&RECEIVER_REPORT_PAYLOAD.to_le_bytes()); + buf.extend_from_slice(&self.timestamp_echo.to_le_bytes()); + buf.extend_from_slice(&self.dwell_time.to_le_bytes()); buf.extend_from_slice(&self.highest_counter.to_le_bytes()); buf.extend_from_slice(&self.cumulative_packets_recv.to_le_bytes()); buf.extend_from_slice(&self.cumulative_bytes_recv.to_le_bytes()); - buf.extend_from_slice(&self.timestamp_echo.to_le_bytes()); - buf.extend_from_slice(&self.dwell_time.to_le_bytes()); - buf.extend_from_slice(&self.max_burst_loss.to_le_bytes()); - buf.extend_from_slice(&self.mean_burst_loss.to_le_bytes()); - buf.extend_from_slice(&[0u8; 2]); // reserved buf.extend_from_slice(&self.jitter.to_le_bytes()); buf.extend_from_slice(&self.ecn_ce_count.to_le_bytes()); buf.extend_from_slice(&self.owd_trend.to_le_bytes()); buf.extend_from_slice(&self.burst_loss_count.to_le_bytes()); buf.extend_from_slice(&self.cumulative_reorder_count.to_le_bytes()); - buf.extend_from_slice(&self.interval_packets_recv.to_le_bytes()); - buf.extend_from_slice(&self.interval_bytes_recv.to_le_bytes()); buf } /// Decode from payload after msg_type byte has been consumed. /// - /// `payload` starts at the reserved bytes (offset 1 in the wire format). + /// `payload` starts at format_version (offset 1 in the wire format). + /// Unknown trailing bytes (from future format extensions) are skipped + /// via total_length. pub fn decode(payload: &[u8]) -> Result { - if payload.len() < 67 { + // Need at least: format_version(1) + total_length(2) + v0 payload(50) = 53 + if payload.len() < 53 { return Err(ProtocolError::MessageTooShort { - expected: 67, + expected: 53, got: payload.len(), }); } - // Skip 3 reserved bytes + let format_version = payload[0]; + let total_length = u16::from_le_bytes(payload[1..3].try_into().unwrap()) as usize; + + if payload.len() < 3 + total_length { + return Err(ProtocolError::MessageTooShort { + expected: 3 + total_length, + got: payload.len(), + }); + } + + if format_version > 0 + && total_length < RECEIVER_REPORT_PAYLOAD as usize + { + return Err(ProtocolError::MessageTooShort { + expected: RECEIVER_REPORT_PAYLOAD as usize, + got: total_length, + }); + } + let p = &payload[3..]; Ok(Self { - highest_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()), - cumulative_packets_recv: u64::from_le_bytes(p[8..16].try_into().unwrap()), - cumulative_bytes_recv: u64::from_le_bytes(p[16..24].try_into().unwrap()), - timestamp_echo: u32::from_le_bytes(p[24..28].try_into().unwrap()), - dwell_time: u16::from_le_bytes(p[28..30].try_into().unwrap()), - max_burst_loss: u16::from_le_bytes(p[30..32].try_into().unwrap()), - mean_burst_loss: u16::from_le_bytes(p[32..34].try_into().unwrap()), - // skip 2 reserved bytes at p[34..36] - jitter: u32::from_le_bytes(p[36..40].try_into().unwrap()), - ecn_ce_count: u32::from_le_bytes(p[40..44].try_into().unwrap()), - owd_trend: i32::from_le_bytes(p[44..48].try_into().unwrap()), - burst_loss_count: u32::from_le_bytes(p[48..52].try_into().unwrap()), - cumulative_reorder_count: u32::from_le_bytes(p[52..56].try_into().unwrap()), - interval_packets_recv: u32::from_le_bytes(p[56..60].try_into().unwrap()), - interval_bytes_recv: u32::from_le_bytes(p[60..64].try_into().unwrap()), + timestamp_echo: u32::from_le_bytes(p[0..4].try_into().unwrap()), + dwell_time: u16::from_le_bytes(p[4..6].try_into().unwrap()), + highest_counter: u64::from_le_bytes(p[6..14].try_into().unwrap()), + cumulative_packets_recv: u64::from_le_bytes(p[14..22].try_into().unwrap()), + cumulative_bytes_recv: u64::from_le_bytes(p[22..30].try_into().unwrap()), + jitter: u32::from_le_bytes(p[30..34].try_into().unwrap()), + ecn_ce_count: u32::from_le_bytes(p[34..38].try_into().unwrap()), + owd_trend: i32::from_le_bytes(p[38..42].try_into().unwrap()), + burst_loss_count: u32::from_le_bytes(p[42..46].try_into().unwrap()), + cumulative_reorder_count: u32::from_le_bytes(p[46..50].try_into().unwrap()), }) } } @@ -181,13 +211,9 @@ use crate::protocol::{SessionReceiverReport, SessionSenderReport}; impl From<&SenderReport> for SessionSenderReport { fn from(r: &SenderReport) -> Self { Self { - interval_start_counter: r.interval_start_counter, - interval_end_counter: r.interval_end_counter, - interval_start_timestamp: r.interval_start_timestamp, - interval_end_timestamp: r.interval_end_timestamp, + interval_packets_sent: r.interval_packets_sent, interval_bytes_sent: r.interval_bytes_sent, cumulative_packets_sent: r.cumulative_packets_sent, - cumulative_bytes_sent: r.cumulative_bytes_sent, } } } @@ -195,13 +221,9 @@ impl From<&SenderReport> for SessionSenderReport { impl From<&SessionSenderReport> for SenderReport { fn from(r: &SessionSenderReport) -> Self { Self { - interval_start_counter: r.interval_start_counter, - interval_end_counter: r.interval_end_counter, - interval_start_timestamp: r.interval_start_timestamp, - interval_end_timestamp: r.interval_end_timestamp, + interval_packets_sent: r.interval_packets_sent, interval_bytes_sent: r.interval_bytes_sent, cumulative_packets_sent: r.cumulative_packets_sent, - cumulative_bytes_sent: r.cumulative_bytes_sent, } } } @@ -209,20 +231,16 @@ impl From<&SessionSenderReport> for SenderReport { impl From<&ReceiverReport> for SessionReceiverReport { fn from(r: &ReceiverReport) -> Self { Self { + timestamp_echo: r.timestamp_echo, + dwell_time: r.dwell_time, highest_counter: r.highest_counter, cumulative_packets_recv: r.cumulative_packets_recv, cumulative_bytes_recv: r.cumulative_bytes_recv, - timestamp_echo: r.timestamp_echo, - dwell_time: r.dwell_time, - max_burst_loss: r.max_burst_loss, - mean_burst_loss: r.mean_burst_loss, jitter: r.jitter, ecn_ce_count: r.ecn_ce_count, owd_trend: r.owd_trend, burst_loss_count: r.burst_loss_count, cumulative_reorder_count: r.cumulative_reorder_count, - interval_packets_recv: r.interval_packets_recv, - interval_bytes_recv: r.interval_bytes_recv, } } } @@ -230,20 +248,16 @@ impl From<&ReceiverReport> for SessionReceiverReport { impl From<&SessionReceiverReport> for ReceiverReport { fn from(r: &SessionReceiverReport) -> Self { Self { + timestamp_echo: r.timestamp_echo, + dwell_time: r.dwell_time, highest_counter: r.highest_counter, cumulative_packets_recv: r.cumulative_packets_recv, cumulative_bytes_recv: r.cumulative_bytes_recv, - timestamp_echo: r.timestamp_echo, - dwell_time: r.dwell_time, - max_burst_loss: r.max_burst_loss, - mean_burst_loss: r.mean_burst_loss, jitter: r.jitter, ecn_ce_count: r.ecn_ce_count, owd_trend: r.owd_trend, burst_loss_count: r.burst_loss_count, cumulative_reorder_count: r.cumulative_reorder_count, - interval_packets_recv: r.interval_packets_recv, - interval_bytes_recv: r.interval_bytes_recv, } } } @@ -258,32 +272,24 @@ mod tests { fn sample_sender_report() -> SenderReport { SenderReport { - interval_start_counter: 100, - interval_end_counter: 200, - interval_start_timestamp: 5000, - interval_end_timestamp: 6000, + interval_packets_sent: 100, interval_bytes_sent: 50_000, cumulative_packets_sent: 10_000, - cumulative_bytes_sent: 5_000_000, } } fn sample_receiver_report() -> ReceiverReport { ReceiverReport { + timestamp_echo: 5900, + dwell_time: 5, highest_counter: 195, cumulative_packets_recv: 9_500, cumulative_bytes_recv: 4_750_000, - timestamp_echo: 5900, - dwell_time: 5, - max_burst_loss: 3, - mean_burst_loss: 384, // 1.5 in u8.8 jitter: 1200, ecn_ce_count: 0, owd_trend: -50, burst_loss_count: 2, cumulative_reorder_count: 10, - interval_packets_recv: 95, - interval_bytes_recv: 47_500, } } @@ -291,15 +297,17 @@ mod tests { fn test_sender_report_encode_size() { let sr = sample_sender_report(); let encoded = sr.encode(); - assert_eq!(encoded.len(), 48); + assert_eq!(encoded.len(), SENDER_REPORT_SIZE); assert_eq!(encoded[0], 0x01); // msg_type + assert_eq!(encoded[1], 0x00); // format_version + let total_len = u16::from_le_bytes([encoded[2], encoded[3]]); + assert_eq!(total_len, SENDER_REPORT_PAYLOAD); } #[test] fn test_sender_report_roundtrip() { let sr = sample_sender_report(); let encoded = sr.encode(); - // decode expects payload after msg_type let decoded = SenderReport::decode(&encoded[1..]).unwrap(); assert_eq!(sr, decoded); } @@ -314,15 +322,17 @@ mod tests { fn test_receiver_report_encode_size() { let rr = sample_receiver_report(); let encoded = rr.encode(); - assert_eq!(encoded.len(), 68); + assert_eq!(encoded.len(), RECEIVER_REPORT_SIZE); assert_eq!(encoded[0], 0x02); // msg_type + assert_eq!(encoded[1], 0x00); // format_version + let total_len = u16::from_le_bytes([encoded[2], encoded[3]]); + assert_eq!(total_len, RECEIVER_REPORT_PAYLOAD); } #[test] fn test_receiver_report_roundtrip() { let rr = sample_receiver_report(); let encoded = rr.encode(); - // decode expects payload after msg_type let decoded = ReceiverReport::decode(&encoded[1..]).unwrap(); assert_eq!(rr, decoded); } @@ -336,13 +346,9 @@ mod tests { #[test] fn test_sender_report_zero_values() { let sr = SenderReport { - interval_start_counter: 0, - interval_end_counter: 0, - interval_start_timestamp: 0, - interval_end_timestamp: 0, + interval_packets_sent: 0, interval_bytes_sent: 0, cumulative_packets_sent: 0, - cumulative_bytes_sent: 0, }; let encoded = sr.encode(); let decoded = SenderReport::decode(&encoded[1..]).unwrap(); @@ -352,20 +358,16 @@ mod tests { #[test] fn test_receiver_report_max_values() { let rr = ReceiverReport { + timestamp_echo: u32::MAX, + dwell_time: u16::MAX, highest_counter: u64::MAX, cumulative_packets_recv: u64::MAX, cumulative_bytes_recv: u64::MAX, - timestamp_echo: u32::MAX, - dwell_time: u16::MAX, - max_burst_loss: u16::MAX, - mean_burst_loss: u16::MAX, jitter: u32::MAX, ecn_ce_count: u32::MAX, owd_trend: i32::MAX, burst_loss_count: u32::MAX, cumulative_reorder_count: u32::MAX, - interval_packets_recv: u32::MAX, - interval_bytes_recv: u32::MAX, }; let encoded = rr.encode(); let decoded = ReceiverReport::decode(&encoded[1..]).unwrap(); @@ -382,4 +384,30 @@ mod tests { let decoded = ReceiverReport::decode(&encoded[1..]).unwrap(); assert_eq!(decoded.owd_trend, -12345); } + + #[test] + fn test_sender_report_forward_compat_trailing_bytes() { + let sr = sample_sender_report(); + let mut encoded = sr.encode(); + // Simulate a future version with extra trailing bytes: + // bump total_length to include 4 extra bytes + let new_total_len = SENDER_REPORT_PAYLOAD + 4; + encoded[2..4].copy_from_slice(&new_total_len.to_le_bytes()); + encoded.extend_from_slice(&[0xAA, 0xBB, 0xCC, 0xDD]); + // Decoder should skip trailing bytes and parse v0 fields + let decoded = SenderReport::decode(&encoded[1..]).unwrap(); + assert_eq!(sr, decoded); + } + + #[test] + fn test_receiver_report_forward_compat_trailing_bytes() { + let rr = sample_receiver_report(); + let mut encoded = rr.encode(); + // Simulate a future version with extra trailing bytes + let new_total_len = RECEIVER_REPORT_PAYLOAD + 8; + encoded[2..4].copy_from_slice(&new_total_len.to_le_bytes()); + encoded.extend_from_slice(&[0x11; 8]); + let decoded = ReceiverReport::decode(&encoded[1..]).unwrap(); + assert_eq!(rr, decoded); + } } diff --git a/src/mmp/sender.rs b/src/mmp/sender.rs index ac7e0a6..9a4a9a0 100644 --- a/src/mmp/sender.rs +++ b/src/mmp/sender.rs @@ -18,16 +18,10 @@ use crate::mmp::{ pub struct SenderState { // --- Cumulative (lifetime) --- cumulative_packets_sent: u64, - cumulative_bytes_sent: u64, // --- Current interval --- - interval_start_counter: u64, - interval_start_timestamp: u32, + interval_packets_sent: u32, interval_bytes_sent: u32, - /// Counter of the most recently sent frame. - last_counter: u64, - /// Timestamp of the most recently sent frame. - last_timestamp: u32, /// Whether any frames have been sent in the current interval. interval_has_data: bool, @@ -56,12 +50,8 @@ impl SenderState { pub fn new_with_cold_start(cold_start_ms: u64) -> Self { Self { cumulative_packets_sent: 0, - cumulative_bytes_sent: 0, - interval_start_counter: 0, - interval_start_timestamp: 0, + interval_packets_sent: 0, interval_bytes_sent: 0, - last_counter: 0, - last_timestamp: 0, interval_has_data: false, last_report_time: None, report_interval: Duration::from_millis(cold_start_ms), @@ -75,17 +65,11 @@ impl SenderState { /// Called on the TX path for every encrypted link message. /// `counter` is the AEAD nonce/counter, `timestamp` is the inner header /// session-relative timestamp (ms), `bytes` is the wire payload size. - pub fn record_sent(&mut self, counter: u64, timestamp: u32, bytes: usize) { - if !self.interval_has_data { - self.interval_start_counter = counter; - self.interval_start_timestamp = timestamp; - self.interval_has_data = true; - } - self.last_counter = counter; - self.last_timestamp = timestamp; + pub fn record_sent(&mut self, _counter: u64, _timestamp: u32, bytes: usize) { + self.interval_has_data = true; + self.interval_packets_sent = self.interval_packets_sent.saturating_add(1); self.interval_bytes_sent = self.interval_bytes_sent.saturating_add(bytes as u32); self.cumulative_packets_sent += 1; - self.cumulative_bytes_sent += bytes as u64; } /// Build a SenderReport from current state and reset the interval. @@ -97,17 +81,14 @@ impl SenderState { } let report = SenderReport { - interval_start_counter: self.interval_start_counter, - interval_end_counter: self.last_counter, - interval_start_timestamp: self.interval_start_timestamp, - interval_end_timestamp: self.last_timestamp, + interval_packets_sent: self.interval_packets_sent, interval_bytes_sent: self.interval_bytes_sent, cumulative_packets_sent: self.cumulative_packets_sent, - cumulative_bytes_sent: self.cumulative_bytes_sent, }; // Reset interval self.interval_has_data = false; + self.interval_packets_sent = 0; self.interval_bytes_sent = 0; self.last_report_time = Some(now); @@ -193,10 +174,6 @@ impl SenderState { self.cumulative_packets_sent } - pub fn cumulative_bytes_sent(&self) -> u64 { - self.cumulative_bytes_sent - } - pub fn report_interval(&self) -> Duration { self.report_interval } @@ -224,7 +201,6 @@ mod tests { fn test_new_sender_state() { let s = SenderState::new(); assert_eq!(s.cumulative_packets_sent(), 0); - assert_eq!(s.cumulative_bytes_sent(), 0); } #[test] @@ -233,7 +209,6 @@ mod tests { s.record_sent(1, 100, 500); s.record_sent(2, 200, 600); assert_eq!(s.cumulative_packets_sent(), 2); - assert_eq!(s.cumulative_bytes_sent(), 1100); } #[test] @@ -250,13 +225,9 @@ mod tests { s.record_sent(12, 1200, 400); let report = s.build_report(Instant::now()).unwrap(); - assert_eq!(report.interval_start_counter, 10); - assert_eq!(report.interval_end_counter, 12); - assert_eq!(report.interval_start_timestamp, 1000); - assert_eq!(report.interval_end_timestamp, 1200); + assert_eq!(report.interval_packets_sent, 3); assert_eq!(report.interval_bytes_sent, 1500); assert_eq!(report.cumulative_packets_sent, 3); - assert_eq!(report.cumulative_bytes_sent, 1500); } #[test] @@ -271,11 +242,10 @@ mod tests { // New data starts a fresh interval s.record_sent(2, 200, 300); let report = s.build_report(Instant::now()).unwrap(); - assert_eq!(report.interval_start_counter, 2); + assert_eq!(report.interval_packets_sent, 1); assert_eq!(report.interval_bytes_sent, 300); // Cumulative continues assert_eq!(report.cumulative_packets_sent, 2); - assert_eq!(report.cumulative_bytes_sent, 800); } #[test] diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 914b23e..375ba65 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -1,8 +1,8 @@ //! Encrypted frame handling (hot path). -use crate::node::Node; -use crate::node::wire::{EncryptedHeader, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, strip_inner_header}; use crate::noise::NoiseError; +use crate::node::Node; +use crate::node::wire::{EncryptedHeader, strip_inner_header, FLAG_CE, FLAG_KEY_EPOCH}; use crate::transport::ReceivedPacket; use std::time::Instant; use tracing::{debug, info, trace, warn}; @@ -53,8 +53,8 @@ impl Node { // Check and perform cutover in a scoped borrow. { let peer = self.peers.get(&node_addr).unwrap(); - let k_bit_flipped = - received_k_bit != peer.current_k_bit() && peer.pending_new_session().is_some(); + let k_bit_flipped = received_k_bit != peer.current_k_bit() + && peer.pending_new_session().is_some(); if k_bit_flipped { let display_name = self.peer_display_name(&node_addr); @@ -70,10 +70,9 @@ impl Node { debug_assert!( peer.transport_id().is_some() && peer.our_index().is_some() - && self.peers_by_index.contains_key(&( - peer.transport_id().unwrap(), - peer.our_index().unwrap().as_u32() - )), + && self.peers_by_index.contains_key( + &(peer.transport_id().unwrap(), peer.our_index().unwrap().as_u32()) + ), "peers_by_index should contain pre-registered new index after K-bit flip" ); } @@ -149,7 +148,6 @@ impl Node { // MMP per-frame processing and statistics let now = Instant::now(); let ce_flag = header.flags & FLAG_CE != 0; - let sp_flag = header.flags & FLAG_SP != 0; if let Some(peer) = self.peers.get_mut(&node_addr) { if let Some(mmp) = peer.mmp_mut() { @@ -160,17 +158,14 @@ impl Node { ce_flag, now, ); - let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now); } peer.set_current_addr(packet.transport_id, packet.remote_addr.clone()); - peer.link_stats_mut() - .record_recv(packet.data.len(), packet.timestamp_ms); + peer.link_stats_mut().record_recv(packet.data.len(), packet.timestamp_ms); peer.touch(packet.timestamp_ms); } // Dispatch to link message handler - self.dispatch_link_message(&node_addr, link_message, ce_flag) - .await; + self.dispatch_link_message(&node_addr, link_message, ce_flag).await; } /// Log a decryption failure with replay suppression. diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index a485542..255ad08 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -59,6 +59,7 @@ impl Node { trace!( from = %self.peer_display_name(from), cum_pkts = sr.cumulative_packets_sent, + interval_pkts = sr.interval_packets_sent, interval_bytes = sr.interval_bytes_sent, "Received SenderReport" ); @@ -290,7 +291,6 @@ impl Node { etx = format_args!("{:.2}", m.etx), goodput = %format_throughput(m.goodput_bps()), tx_pkts = mmp.sender.cumulative_packets_sent(), - tx_bytes = mmp.sender.cumulative_bytes_sent(), rx_pkts = mmp.receiver.cumulative_packets_recv(), rx_bytes = mmp.receiver.cumulative_bytes_recv(), "MMP link teardown" @@ -489,7 +489,6 @@ impl Node { send_mtu = mmp.path_mtu.current_mtu(), observed_mtu = mmp.path_mtu.last_observed_mtu(), tx_pkts = mmp.sender.cumulative_packets_sent(), - tx_bytes = mmp.sender.cumulative_bytes_sent(), rx_pkts = mmp.receiver.cumulative_packets_recv(), rx_bytes = mmp.receiver.cumulative_bytes_recv(), "MMP session teardown" diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 94a0c2e..bf90b79 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -251,13 +251,7 @@ impl Node { mmp.receiver.record_recv( header.counter, timestamp, plaintext.len(), ce_flag, now, ); - // Spin bit: advance state machine for correct TX reflection. - // RTT samples not fed into SRTT — timestamp-echo provides - // accurate RTT; spin bit includes variable inter-frame delays. - let inner_flags = FspInnerFlags::from_byte(inner_flags_byte); - let _spin_rtt = mmp.spin_bit.rx_observe( - inner_flags.spin_bit, header.counter, now, - ); + let _inner_flags = FspInnerFlags::from_byte(inner_flags_byte); } // Feed path_mtu from datagram envelope to MMP path MTU tracking. @@ -915,6 +909,7 @@ impl Node { trace!( src = %self.peer_display_name(src_addr), cum_pkts = sr.cumulative_packets_sent, + interval_pkts = sr.interval_packets_sent, interval_bytes = sr.interval_bytes_sent, "Received SessionSenderReport" ); @@ -1273,7 +1268,6 @@ impl Node { })?; let wants_coords = entry.coords_warmup_remaining() > 0; let timestamp = entry.session_timestamp(now_ms); - let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit()); if !entry.is_established() { return Err(NodeError::SendFailed { node_addr: *dest_addr, @@ -1289,7 +1283,7 @@ impl Node { // Build inner plaintext (doesn't depend on counter) let msg_type = SessionMessageType::DataPacket.to_byte(); // 0x10 - let inner_flags = FspInnerFlags { spin_bit }.to_byte(); + let inner_flags = FspInnerFlags::new().to_byte(); let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, &port_payload); // Determine whether coords fit within transport MTU. @@ -1422,10 +1416,8 @@ impl Node { reason: "no session".into(), })?; let timestamp = entry.session_timestamp(now_ms); - let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit()); - // Build inner flags with spin bit - let inner_flags = FspInnerFlags { spin_bit }.to_byte(); + let inner_flags = FspInnerFlags::new().to_byte(); // Get mutable access for encryption let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed { @@ -1506,7 +1498,6 @@ impl Node { reason: "no session".into(), })?; let timestamp = entry.session_timestamp(now_ms); - let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit()); // Get mutable access for encryption let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed { @@ -1527,7 +1518,7 @@ impl Node { // FSP inner header only, no body payload let msg_type = SessionMessageType::CoordsWarmup.to_byte(); - let inner_flags = FspInnerFlags { spin_bit }.to_byte(); + let inner_flags = FspInnerFlags::new().to_byte(); let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, &[]); // Build FSP header with CP flag diff --git a/src/node/mod.rs b/src/node/mod.rs index fbf3283..f8b384c 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -40,7 +40,7 @@ use crate::tree::TreeState; use crate::upper::hosts::HostMap; use crate::upper::icmp_rate_limit::IcmpRateLimiter; use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; -use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP}; +use self::wire::{build_encrypted, build_established_header, prepend_inner_header, FLAG_CE, FLAG_KEY_EPOCH}; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity}; use rand::Rng; use std::collections::{HashMap, VecDeque}; @@ -1637,11 +1637,7 @@ impl Node { // Prepend 4-byte session-relative timestamp (inner header) let timestamp_ms = peer.session_elapsed_ms(); - // MMP: read spin bit value before entering session borrow - let sp_flag = peer.mmp() - .map(|mmp| mmp.spin_bit.tx_bit()) - .unwrap_or(false); - let mut flags = if sp_flag { FLAG_SP } else { 0 }; + let mut flags = 0u8; if ce_flag { flags |= FLAG_CE; } diff --git a/src/node/session_wire.rs b/src/node/session_wire.rs index 7cff06f..f675872 100644 --- a/src/node/session_wire.rs +++ b/src/node/session_wire.rs @@ -89,11 +89,6 @@ pub const FSP_FLAG_K: u8 = 0x02; /// Unencrypted — payload is plaintext (error signals). pub const FSP_FLAG_U: u8 = 0x04; -// Inner flag bit constants (byte 5 of decrypted inner header). - -/// Spin bit for end-to-end RTT measurement (inside AEAD). -#[allow(dead_code)] -pub const FSP_INNER_FLAG_SP: u8 = 0x01; // ============================================================================ // Common Prefix diff --git a/src/node/wire.rs b/src/node/wire.rs index d91acb3..a55ab48 100644 --- a/src/node/wire.rs +++ b/src/node/wire.rs @@ -71,9 +71,6 @@ pub const FLAG_KEY_EPOCH: u8 = 0x01; #[allow(dead_code)] /// Congestion Experienced echo flag. pub const FLAG_CE: u8 = 0x02; -#[allow(dead_code)] -/// Spin bit for RTT measurement. -pub const FLAG_SP: u8 = 0x04; // ============================================================================ // Common Prefix @@ -516,11 +513,11 @@ mod tests { #[test] fn test_common_prefix_parse() { - let data = [0x10, 0x04, 0x20, 0x00]; // ver=1, phase=0, flags=SP, payload_len=32 + let data = [0x10, 0x02, 0x20, 0x00]; // ver=1, phase=0, flags=CE, payload_len=32 let prefix = CommonPrefix::parse(&data).unwrap(); assert_eq!(prefix.version, 1); assert_eq!(prefix.phase, 0); - assert_eq!(prefix.flags, FLAG_SP); + assert_eq!(prefix.flags, FLAG_CE); assert_eq!(prefix.payload_len, 32); } @@ -698,10 +695,10 @@ mod tests { let header = build_established_header( SessionIndex::new(1), 0, - FLAG_KEY_EPOCH | FLAG_SP, + FLAG_KEY_EPOCH | FLAG_CE, 100, ); - assert_eq!(header[1], 0x05); // bits 0 and 2 set + assert_eq!(header[1], 0x03); // bits 0 and 1 set let parsed = EncryptedHeader::parse(&[ header[0], header[1], header[2], header[3], @@ -712,8 +709,7 @@ mod tests { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, ]).unwrap(); assert_eq!(parsed.flags & FLAG_KEY_EPOCH, FLAG_KEY_EPOCH); - assert_eq!(parsed.flags & FLAG_CE, 0); - assert_eq!(parsed.flags & FLAG_SP, FLAG_SP); + assert_eq!(parsed.flags & FLAG_CE, FLAG_CE); } #[test] diff --git a/src/protocol/session.rs b/src/protocol/session.rs index b28365c..7af0584 100644 --- a/src/protocol/session.rs +++ b/src/protocol/session.rs @@ -295,30 +295,24 @@ impl FspFlags { /// /// | Bit | Name | Description | /// |-----|------|---------------------------------| -/// | 0 | SP | Spin bit for RTT measurement | -/// | 1-7 | | Reserved | +/// | 0-7 | | Reserved (all zero) | #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub struct FspInnerFlags { - /// Spin bit for passive RTT measurement. - pub spin_bit: bool, -} +pub struct FspInnerFlags; impl FspInnerFlags { /// Create default inner flags (all clear). pub fn new() -> Self { - Self::default() + Self } /// Convert to a byte. pub fn to_byte(&self) -> u8 { - if self.spin_bit { 0x01 } else { 0x00 } + 0x00 } /// Convert from a byte. - pub fn from_byte(byte: u8) -> Self { - Self { - spin_bit: byte & 0x01 != 0, - } + pub fn from_byte(_byte: u8) -> Self { + Self } } @@ -664,47 +658,40 @@ impl SessionMsg3 { /// Session-layer sender report (msg_type 0x11). /// /// Mirrors the FMP `SenderReport` fields but carried as an FSP session -/// message inside the AEAD envelope. The msg_type is in the FSP inner -/// header, so the body starts with reserved bytes. +/// message inside the AEAD envelope. Uses the same extensibility header +/// as the link-layer format: `[format_version:1][total_length:2 LE]`. /// -/// ## Wire Format (46 bytes body, after inner header stripped) +/// ## Wire Format (19 bytes body, after inner header stripped) /// /// ```text -/// [0-1] reserved (zero) -/// [2-9] interval_start_counter: u64 LE -/// [10-17] interval_end_counter: u64 LE -/// [18-21] interval_start_timestamp: u32 LE -/// [22-25] interval_end_timestamp: u32 LE -/// [26-29] interval_bytes_sent: u32 LE -/// [30-37] cumulative_packets_sent: u64 LE -/// [38-45] cumulative_bytes_sent: u64 LE +/// [0] format_version = 0 +/// [1-2] total_length: u16 LE (= 16) +/// [3-6] interval_packets_sent: u32 LE +/// [7-10] interval_bytes_sent: u32 LE +/// [11-18] cumulative_packets_sent: u64 LE /// ``` #[derive(Debug, Clone, PartialEq, Eq)] pub struct SessionSenderReport { - pub interval_start_counter: u64, - pub interval_end_counter: u64, - pub interval_start_timestamp: u32, - pub interval_end_timestamp: u32, + pub interval_packets_sent: u32, pub interval_bytes_sent: u32, pub cumulative_packets_sent: u64, - pub cumulative_bytes_sent: u64, } -/// Body size for SessionSenderReport: 2 reserved + 44 fields. -pub const SESSION_SENDER_REPORT_SIZE: usize = 46; +/// Body size for SessionSenderReport: format_version(1) + total_length(2) + payload(16). +pub const SESSION_SENDER_REPORT_SIZE: usize = 19; + +/// Payload size after total_length field for SessionSenderReport format v0. +const SESSION_SR_PAYLOAD: u16 = 16; impl SessionSenderReport { - /// Encode to wire format (46 bytes body). + /// Encode to wire format (19 bytes body). pub fn encode(&self) -> Vec { let mut buf = Vec::with_capacity(SESSION_SENDER_REPORT_SIZE); - buf.extend_from_slice(&[0u8; 2]); // reserved - buf.extend_from_slice(&self.interval_start_counter.to_le_bytes()); - buf.extend_from_slice(&self.interval_end_counter.to_le_bytes()); - buf.extend_from_slice(&self.interval_start_timestamp.to_le_bytes()); - buf.extend_from_slice(&self.interval_end_timestamp.to_le_bytes()); + buf.push(0x00); // format_version + buf.extend_from_slice(&SESSION_SR_PAYLOAD.to_le_bytes()); + buf.extend_from_slice(&self.interval_packets_sent.to_le_bytes()); buf.extend_from_slice(&self.interval_bytes_sent.to_le_bytes()); buf.extend_from_slice(&self.cumulative_packets_sent.to_le_bytes()); - buf.extend_from_slice(&self.cumulative_bytes_sent.to_le_bytes()); buf } @@ -716,16 +703,19 @@ impl SessionSenderReport { got: body.len(), }); } - // Skip 2 reserved bytes - let p = &body[2..]; + let _format_version = body[0]; + let total_length = u16::from_le_bytes(body[1..3].try_into().unwrap()) as usize; + if body.len() < 3 + total_length { + return Err(ProtocolError::MessageTooShort { + expected: 3 + total_length, + got: body.len(), + }); + } + let p = &body[3..]; Ok(Self { - interval_start_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()), - interval_end_counter: u64::from_le_bytes(p[8..16].try_into().unwrap()), - interval_start_timestamp: u32::from_le_bytes(p[16..20].try_into().unwrap()), - interval_end_timestamp: u32::from_le_bytes(p[20..24].try_into().unwrap()), - interval_bytes_sent: u32::from_le_bytes(p[24..28].try_into().unwrap()), - cumulative_packets_sent: u64::from_le_bytes(p[28..36].try_into().unwrap()), - cumulative_bytes_sent: u64::from_le_bytes(p[36..44].try_into().unwrap()), + interval_packets_sent: u32::from_le_bytes(p[0..4].try_into().unwrap()), + interval_bytes_sent: u32::from_le_bytes(p[4..8].try_into().unwrap()), + cumulative_packets_sent: u64::from_le_bytes(p[8..16].try_into().unwrap()), }) } } @@ -733,69 +723,61 @@ impl SessionSenderReport { /// Session-layer receiver report (msg_type 0x12). /// /// Mirrors the FMP `ReceiverReport` fields but carried as an FSP session -/// message inside the AEAD envelope. +/// message inside the AEAD envelope. Uses the same extensibility header +/// as the link-layer format: `[format_version:1][total_length:2 LE]`. /// -/// ## Wire Format (66 bytes body, after inner header stripped) +/// ## Wire Format (53 bytes body, after inner header stripped) /// /// ```text -/// [0-1] reserved (zero) -/// [2-9] highest_counter: u64 LE -/// [10-17] cumulative_packets_recv: u64 LE -/// [18-25] cumulative_bytes_recv: u64 LE -/// [26-29] timestamp_echo: u32 LE -/// [30-31] dwell_time: u16 LE -/// [32-33] max_burst_loss: u16 LE -/// [34-35] mean_burst_loss: u16 LE (u8.8 fixed-point) -/// [36-37] reserved: u16 LE -/// [38-41] jitter: u32 LE (microseconds) -/// [42-45] ecn_ce_count: u32 LE -/// [46-49] owd_trend: i32 LE (µs/s) -/// [50-53] burst_loss_count: u32 LE -/// [54-57] cumulative_reorder_count: u32 LE -/// [58-61] interval_packets_recv: u32 LE -/// [62-65] interval_bytes_recv: u32 LE +/// [0] format_version = 0 +/// [1-2] total_length: u16 LE (= 50) +/// [3-6] timestamp_echo: u32 LE +/// [7-8] dwell_time: u16 LE +/// [9-16] highest_counter: u64 LE +/// [17-24] cumulative_packets_recv: u64 LE +/// [25-32] cumulative_bytes_recv: u64 LE +/// [33-36] jitter: u32 LE (microseconds) +/// [37-40] ecn_ce_count: u32 LE +/// [41-44] owd_trend: i32 LE (µs/s) +/// [45-48] burst_loss_count: u32 LE +/// [49-52] cumulative_reorder_count: u32 LE /// ``` #[derive(Debug, Clone, PartialEq, Eq)] pub struct SessionReceiverReport { + pub timestamp_echo: u32, + pub dwell_time: u16, pub highest_counter: u64, pub cumulative_packets_recv: u64, pub cumulative_bytes_recv: u64, - pub timestamp_echo: u32, - pub dwell_time: u16, - pub max_burst_loss: u16, - pub mean_burst_loss: u16, pub jitter: u32, pub ecn_ce_count: u32, pub owd_trend: i32, pub burst_loss_count: u32, pub cumulative_reorder_count: u32, - pub interval_packets_recv: u32, - pub interval_bytes_recv: u32, } -/// Body size for SessionReceiverReport: 2 reserved + 64 fields. -pub const SESSION_RECEIVER_REPORT_SIZE: usize = 66; +/// Body size for SessionReceiverReport: format_version(1) + total_length(2) + payload(50). +pub const SESSION_RECEIVER_REPORT_SIZE: usize = 53; + +/// Payload size after total_length field for SessionReceiverReport format v0. +const SESSION_RR_PAYLOAD: u16 = 50; impl SessionReceiverReport { - /// Encode to wire format (66 bytes body). + /// Encode to wire format (53 bytes body). pub fn encode(&self) -> Vec { let mut buf = Vec::with_capacity(SESSION_RECEIVER_REPORT_SIZE); - buf.extend_from_slice(&[0u8; 2]); // reserved + buf.push(0x00); // format_version + buf.extend_from_slice(&SESSION_RR_PAYLOAD.to_le_bytes()); + buf.extend_from_slice(&self.timestamp_echo.to_le_bytes()); + buf.extend_from_slice(&self.dwell_time.to_le_bytes()); buf.extend_from_slice(&self.highest_counter.to_le_bytes()); buf.extend_from_slice(&self.cumulative_packets_recv.to_le_bytes()); buf.extend_from_slice(&self.cumulative_bytes_recv.to_le_bytes()); - buf.extend_from_slice(&self.timestamp_echo.to_le_bytes()); - buf.extend_from_slice(&self.dwell_time.to_le_bytes()); - buf.extend_from_slice(&self.max_burst_loss.to_le_bytes()); - buf.extend_from_slice(&self.mean_burst_loss.to_le_bytes()); - buf.extend_from_slice(&[0u8; 2]); // reserved buf.extend_from_slice(&self.jitter.to_le_bytes()); buf.extend_from_slice(&self.ecn_ce_count.to_le_bytes()); buf.extend_from_slice(&self.owd_trend.to_le_bytes()); buf.extend_from_slice(&self.burst_loss_count.to_le_bytes()); buf.extend_from_slice(&self.cumulative_reorder_count.to_le_bytes()); - buf.extend_from_slice(&self.interval_packets_recv.to_le_bytes()); - buf.extend_from_slice(&self.interval_bytes_recv.to_le_bytes()); buf } @@ -807,24 +789,26 @@ impl SessionReceiverReport { got: body.len(), }); } - // Skip 2 reserved bytes - let p = &body[2..]; + let _format_version = body[0]; + let total_length = u16::from_le_bytes(body[1..3].try_into().unwrap()) as usize; + if body.len() < 3 + total_length { + return Err(ProtocolError::MessageTooShort { + expected: 3 + total_length, + got: body.len(), + }); + } + let p = &body[3..]; Ok(Self { - highest_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()), - cumulative_packets_recv: u64::from_le_bytes(p[8..16].try_into().unwrap()), - cumulative_bytes_recv: u64::from_le_bytes(p[16..24].try_into().unwrap()), - timestamp_echo: u32::from_le_bytes(p[24..28].try_into().unwrap()), - dwell_time: u16::from_le_bytes(p[28..30].try_into().unwrap()), - max_burst_loss: u16::from_le_bytes(p[30..32].try_into().unwrap()), - mean_burst_loss: u16::from_le_bytes(p[32..34].try_into().unwrap()), - // skip 2 reserved bytes at p[34..36] - jitter: u32::from_le_bytes(p[36..40].try_into().unwrap()), - ecn_ce_count: u32::from_le_bytes(p[40..44].try_into().unwrap()), - owd_trend: i32::from_le_bytes(p[44..48].try_into().unwrap()), - burst_loss_count: u32::from_le_bytes(p[48..52].try_into().unwrap()), - cumulative_reorder_count: u32::from_le_bytes(p[52..56].try_into().unwrap()), - interval_packets_recv: u32::from_le_bytes(p[56..60].try_into().unwrap()), - interval_bytes_recv: u32::from_le_bytes(p[60..64].try_into().unwrap()), + timestamp_echo: u32::from_le_bytes(p[0..4].try_into().unwrap()), + dwell_time: u16::from_le_bytes(p[4..6].try_into().unwrap()), + highest_counter: u64::from_le_bytes(p[6..14].try_into().unwrap()), + cumulative_packets_recv: u64::from_le_bytes(p[14..22].try_into().unwrap()), + cumulative_bytes_recv: u64::from_le_bytes(p[22..30].try_into().unwrap()), + jitter: u32::from_le_bytes(p[30..34].try_into().unwrap()), + ecn_ce_count: u32::from_le_bytes(p[34..38].try_into().unwrap()), + owd_trend: i32::from_le_bytes(p[38..42].try_into().unwrap()), + burst_loss_count: u32::from_le_bytes(p[42..46].try_into().unwrap()), + cumulative_reorder_count: u32::from_le_bytes(p[46..50].try_into().unwrap()), }) } } @@ -1430,30 +1414,14 @@ mod tests { #[test] fn test_fsp_inner_flags_default() { let flags = FspInnerFlags::new(); - assert!(!flags.spin_bit); assert_eq!(flags.to_byte(), 0x00); } #[test] - fn test_fsp_inner_flags_roundtrip() { - let flags = FspInnerFlags::from_byte(0x01); - assert!(flags.spin_bit); - assert_eq!(flags.to_byte(), 0x01); - - let flags = FspInnerFlags::from_byte(0x00); - assert!(!flags.spin_bit); - assert_eq!(flags.to_byte(), 0x00); - } - - #[test] - fn test_fsp_inner_flags_ignores_reserved() { - let flags = FspInnerFlags::from_byte(0xFE); - assert!(!flags.spin_bit); - assert_eq!(flags.to_byte(), 0x00); - + fn test_fsp_inner_flags_from_byte() { + // All bits are reserved; from_byte always returns default let flags = FspInnerFlags::from_byte(0xFF); - assert!(flags.spin_bit); - assert_eq!(flags.to_byte(), 0x01); + assert_eq!(flags.to_byte(), 0x00); } // ===== New SessionMessageType Values ===== @@ -1485,13 +1453,9 @@ mod tests { fn sample_session_sender_report() -> SessionSenderReport { SessionSenderReport { - interval_start_counter: 100, - interval_end_counter: 200, - interval_start_timestamp: 5000, - interval_end_timestamp: 6000, + interval_packets_sent: 100, interval_bytes_sent: 50_000, cumulative_packets_sent: 10_000, - cumulative_bytes_sent: 5_000_000, } } @@ -1519,20 +1483,16 @@ mod tests { fn sample_session_receiver_report() -> SessionReceiverReport { SessionReceiverReport { + timestamp_echo: 5900, + dwell_time: 5, highest_counter: 195, cumulative_packets_recv: 9_500, cumulative_bytes_recv: 4_750_000, - timestamp_echo: 5900, - dwell_time: 5, - max_burst_loss: 3, - mean_burst_loss: 384, jitter: 1200, ecn_ce_count: 0, owd_trend: -50, burst_loss_count: 2, cumulative_reorder_count: 10, - interval_packets_recv: 95, - interval_bytes_recv: 47_500, } }