laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-01 04:26:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove spin bit, slim down MMP reports with extensibility header

Browse Source 
Delete SpinBitState, FLAG_SP (FMP bit 2), and FspInnerFlags.spin_bit.
Spin bit superseded by MMP receiver report timestamp echo for RTT.
Reclaims FMP flags bit 2 and FSP inner flags bit 0.

SenderReport: 48 -> 20 bytes. Three fields: interval_packets_sent,
interval_bytes_sent, cumulative_packets_sent.

ReceiverReport: 68 -> 54 bytes. Ten decision-driving and diagnostic
fields retained; removed max_burst_loss, mean_burst_loss,
interval_packets_recv, interval_bytes_recv.

Both report types use extensibility header: repurposed 3 reserved
bytes as [format_version:1][total_length:2 LE]. Decoders skip
unknown trailing bytes for forward compatibility. Session-layer
reports updated to match.
This commit is contained in:
Johnathan Corgan
2026-04-11 08:16:01 +00:00
parent f4278f1dd7
commit 4aded9a238
14 changed files with 296 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/control/queries.rs
-1
View File
@@ -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() {
src/mmp/algorithms.rs
-120
View File
@@ -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<Instant>,
}
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<std::time::Duration> {
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
}
}
src/mmp/metrics.rs
+2 -6
View File
@@ -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,
}
}
src/mmp/mod.rs
+5 -13
View File
@@ -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<Instant>,
@@ -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<Instant>,
@@ -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,
src/mmp/receiver.rs
+3 -11
View File
@@ -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);
}
src/mmp/report.rs
+167 -139
View File
@@ -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<u8> {
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<Self, ProtocolError> {
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<u8> {
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<Self, ProtocolError> {
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);
}
}
src/mmp/sender.rs
+9 -39
View File
@@ -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]
src/node/handlers/encrypted.rs
+9 -14
View File
@@ -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.
src/node/handlers/mmp.rs
+1 -2
View File
@@ -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"
src/node/handlers/session.rs
+5 -14
View File
@@ -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
src/node/mod.rs
+2 -6
View File
@@ -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;
}
src/node/session_wire.rs
-5
View File
@@ -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
src/node/wire.rs
+5 -9
View File
@@ -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]
src/protocol/session.rs
+88 -128
View File
@@ -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<u8> {
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<u8> {
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,
}
}