laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-30 19:46:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

proto/mmp: split state into per-role modules and dissolve the vestigial mmp shell

Browse Source 
Reorganize the MMP subsystem, behavior unchanged throughout:

- Restore the pre-migration role split: move SenderState into sender.rs,
  ReceiverState (with its GapTracker helper) into receiver.rs, MmpMetrics and
  RrLog into metrics.rs, and PathMtuState into path_mtu.rs, leaving the Mmp
  aggregate plus the peer/session state in state.rs (1339 to 196 lines). Home
  the module constants in a new limits.rs and re-export them at the same paths.
  Pure code-motion with module rewiring; visibility unchanged.

- Dissolve the 97-line src/mmp shell, which held only MmpConfig and the
  monotonic mono_ms() clock: move MmpConfig into config/node.rs (re-exported as
  crate::config::MmpConfig), move mono_ms() into a new top-level src/time.rs
  shell time seam, repoint all callers, and delete src/mmp/. Also correct stale
  src/mmp/ doc-comment path labels to proto/mmp/ where they name the protocol
  primitives.
This commit is contained in:
Johnathan Corgan
2026-07-08 19:00:24 +00:00
parent c1ddbf053c
commit 309a91d293
23 changed files with 1380 additions and 1347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/mod.rs
+2 -2
View File
@@ -34,8 +34,8 @@ use thiserror::Error;
pub use gateway::{ConntrackConfig, GatewayConfig, GatewayDnsConfig, PortForward, Proto};
pub use node::{
BloomConfig, BuffersConfig, CacheConfig, ControlConfig, DiscoveryConfig, LimitsConfig,
NodeConfig, NostrDiscoveryConfig, NostrDiscoveryPolicy, RateLimitConfig, RekeyConfig,
RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig,
MmpConfig, NodeConfig, NostrDiscoveryConfig, NostrDiscoveryPolicy, RateLimitConfig,
RekeyConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig,
};
pub use peer::{ConnectPolicy, PeerAddress, PeerConfig};
pub use transport::{
src/config/node.rs
+65 -1
View File
@@ -7,7 +7,6 @@
use serde::{Deserialize, Serialize};
use super::IdentityConfig;
use crate::mmp::MmpConfig;
use crate::proto::mmp::{DEFAULT_LOG_INTERVAL_SECS, DEFAULT_OWD_WINDOW_SIZE, MmpMode};
// ============================================================================
@@ -727,6 +726,41 @@ impl SessionConfig {
}
}
/// MMP configuration (`node.mmp.*`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MmpConfig {
/// Operating mode (`node.mmp.mode`).
#[serde(default)]
pub mode: MmpMode,
/// Periodic operator log interval in seconds (`node.mmp.log_interval_secs`).
#[serde(default = "MmpConfig::default_log_interval_secs")]
pub log_interval_secs: u64,
/// OWD trend ring buffer size (`node.mmp.owd_window_size`).
#[serde(default = "MmpConfig::default_owd_window_size")]
pub owd_window_size: usize,
}
impl Default for MmpConfig {
fn default() -> Self {
Self {
mode: MmpMode::default(),
log_interval_secs: DEFAULT_LOG_INTERVAL_SECS,
owd_window_size: DEFAULT_OWD_WINDOW_SIZE,
}
}
}
impl MmpConfig {
fn default_log_interval_secs() -> u64 {
DEFAULT_LOG_INTERVAL_SECS
}
fn default_owd_window_size() -> usize {
DEFAULT_OWD_WINDOW_SIZE
}
}
/// Session-layer Metrics Measurement Protocol (`node.session_mmp.*`).
///
/// Separate from link-layer `node.mmp.*` to allow independent mode/interval
@@ -1096,6 +1130,36 @@ impl NodeConfig {
mod tests {
use super::*;
#[test]
fn test_config_default() {
let config = MmpConfig::default();
assert_eq!(config.mode, MmpMode::Full);
assert_eq!(config.log_interval_secs, 30);
assert_eq!(config.owd_window_size, 32);
}
#[test]
fn test_config_yaml_parse() {
let yaml = r#"
mode: lightweight
log_interval_secs: 60
owd_window_size: 48
"#;
let config: MmpConfig = serde_yaml::from_str(yaml).unwrap();
assert_eq!(config.mode, MmpMode::Lightweight);
assert_eq!(config.log_interval_secs, 60);
assert_eq!(config.owd_window_size, 48);
}
#[test]
fn test_config_yaml_partial() {
let yaml = "mode: minimal";
let config: MmpConfig = serde_yaml::from_str(yaml).unwrap();
assert_eq!(config.mode, MmpMode::Minimal);
assert_eq!(config.log_interval_secs, DEFAULT_LOG_INTERVAL_SECS);
assert_eq!(config.owd_window_size, DEFAULT_OWD_WINDOW_SIZE);
}
#[test]
fn test_ecn_config_defaults() {
let c = EcnConfig::default();
src/lib.rs
+1 -1
View File
@@ -16,7 +16,6 @@ pub mod discovery;
#[cfg(target_os = "linux")]
pub mod gateway;
pub mod identity;
pub mod mmp;
pub mod node;
pub mod noise;
pub mod peer;
@@ -24,6 +23,7 @@ pub mod perf_profile;
pub(crate) mod proto;
#[cfg(test)]
pub(crate) mod testutil;
mod time;
pub mod transport;
pub mod upper;
pub mod utils;
src/mmp/mod.rs
-97
View File
@@ -1,97 +0,0 @@
//! Metrics Measurement Protocol (MMP) — node-side shell.
//!
//! The MMP protocol core (estimators, sender/receiver/metrics/path-MTU state
//! machines, report codecs and the reporting decisions) lives in the sans-IO
//! [`crate::proto::mmp`] module. What remains here is the shell-only surface:
//! the serde node configuration ([`MmpConfig`]) and the monotonic millisecond
//! clock ([`mono_ms`]) the shell reads at the edge and injects into the owned
//! `u64`-ms state.
use std::sync::LazyLock;
use std::time::Instant;
use serde::{Deserialize, Serialize};
use crate::proto::mmp::{DEFAULT_LOG_INTERVAL_SECS, DEFAULT_OWD_WINDOW_SIZE, MmpMode};
/// Monotonic millisecond clock for the MMP time seam.
///
/// The owned `proto::mmp` state takes injected `u64` milliseconds rather than
/// reading a clock. The shell reads this process-monotonic clock at each edge
/// and passes the value in; only deltas between two `mono_ms()` reads are ever
/// compared, so the (process-relative) epoch is immaterial.
pub(crate) fn mono_ms() -> u64 {
static START: LazyLock<Instant> = LazyLock::new(Instant::now);
START.elapsed().as_millis() as u64
}
/// MMP configuration (`node.mmp.*`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MmpConfig {
/// Operating mode (`node.mmp.mode`).
#[serde(default)]
pub mode: MmpMode,
/// Periodic operator log interval in seconds (`node.mmp.log_interval_secs`).
#[serde(default = "MmpConfig::default_log_interval_secs")]
pub log_interval_secs: u64,
/// OWD trend ring buffer size (`node.mmp.owd_window_size`).
#[serde(default = "MmpConfig::default_owd_window_size")]
pub owd_window_size: usize,
}
impl Default for MmpConfig {
fn default() -> Self {
Self {
mode: MmpMode::default(),
log_interval_secs: DEFAULT_LOG_INTERVAL_SECS,
owd_window_size: DEFAULT_OWD_WINDOW_SIZE,
}
}
}
impl MmpConfig {
fn default_log_interval_secs() -> u64 {
DEFAULT_LOG_INTERVAL_SECS
}
fn default_owd_window_size() -> usize {
DEFAULT_OWD_WINDOW_SIZE
}
}
#[cfg(test)]
mod tests {
use super::MmpConfig;
use crate::proto::mmp::{DEFAULT_LOG_INTERVAL_SECS, DEFAULT_OWD_WINDOW_SIZE, MmpMode};
#[test]
fn test_config_default() {
let config = MmpConfig::default();
assert_eq!(config.mode, MmpMode::Full);
assert_eq!(config.log_interval_secs, 30);
assert_eq!(config.owd_window_size, 32);
}
#[test]
fn test_config_yaml_parse() {
let yaml = r#"
mode: lightweight
log_interval_secs: 60
owd_window_size: 48
"#;
let config: MmpConfig = serde_yaml::from_str(yaml).unwrap();
assert_eq!(config.mode, MmpMode::Lightweight);
assert_eq!(config.log_interval_secs, 60);
assert_eq!(config.owd_window_size, 48);
}
#[test]
fn test_config_yaml_partial() {
let yaml = "mode: minimal";
let config: MmpConfig = serde_yaml::from_str(yaml).unwrap();
assert_eq!(config.mode, MmpMode::Minimal);
assert_eq!(config.log_interval_secs, DEFAULT_LOG_INTERVAL_SECS);
assert_eq!(config.owd_window_size, DEFAULT_OWD_WINDOW_SIZE);
}
}
src/node/handlers/encrypted.rs
+2 -2
View File
@@ -258,7 +258,7 @@ impl Node {
};
// MMP per-frame processing and statistics
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
let ce_flag = header.flags & FLAG_CE != 0;
let sp_flag = header.flags & FLAG_SP != 0;
@@ -354,7 +354,7 @@ impl Node {
} else {
return;
};
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
let mut address_changed = false;
if let Some(peer) = self.peers.get_mut(node_addr) {
peer.reset_decrypt_failures();
src/node/handlers/mmp.rs
+4 -4
View File
@@ -152,7 +152,7 @@ impl Node {
// Process the report: computes RTT from timestamp echo, updates
// loss rate, goodput rate, jitter trend, and ETX.
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
let (first_rtt, rr_log) =
mmp.metrics
.process_receiver_report(&rr, our_timestamp_ms, now_ms);
@@ -197,7 +197,7 @@ impl Node {
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let mono_now_ms = crate::mmp::mono_ms();
let mono_now_ms = crate::time::mono_ms();
if let Some(new_parent) = self.tree_state.evaluate_parent(
&peer_costs,
&std::collections::BTreeSet::new(),
@@ -277,7 +277,7 @@ impl Node {
///
/// Called from the tick handler. Also emits periodic operator logs.
pub(in crate::node) async fn check_mmp_reports(&mut self) {
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
// Build one report-gating snapshot per peer, resolving every timing read
// shell-side into a `bool`. `send_sr`/`send_rr` are `true` on the master
@@ -427,7 +427,7 @@ impl Node {
// Monotonic ms for the MMP receiver's injected-`u64` liveness clock; the
// Instant `now` is still used for the shell-owned heartbeat timing and
// the session-start fallback (both `ActivePeer` Instants).
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
let heartbeat_interval = Duration::from_secs(self.config().node.heartbeat_interval_secs);
let dead_timeout = Duration::from_secs(self.config().node.link_dead_timeout_secs);
let dead_timeout_ms = dead_timeout.as_millis() as u64;
src/node/handlers/session.rs
+5 -5
View File
@@ -324,7 +324,7 @@ impl Node {
if let Some(entry) = self.sessions.get_mut(src_addr)
&& let Some(mmp) = entry.mmp_mut()
{
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
mmp.receiver
.record_recv(header.counter, timestamp, plaintext.len(), ce_flag, now_ms);
// Spin bit: advance state machine for correct TX reflection.
@@ -946,7 +946,7 @@ impl Node {
/// Called from the tick handler. Also emits periodic session MMP logs.
/// Uses the collect-then-send pattern to avoid borrowing conflicts.
pub(in crate::node) async fn check_session_mmp_reports(&mut self) {
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
// Build one report-gating snapshot per session, resolving every timing
// read shell-side into a `bool`. The snapshots own only
@@ -1213,7 +1213,7 @@ impl Node {
let (_first_rtt, rr_log) =
mmp.metrics
.process_receiver_report(&rr, our_timestamp_ms, crate::mmp::mono_ms());
.process_receiver_report(&rr, our_timestamp_ms, crate::time::mono_ms());
// Re-emit the operator trace the core used to log mid-decision.
super::mmp::log_rr_outcome(&rr, our_timestamp_ms, rr_log);
@@ -1283,7 +1283,7 @@ impl Node {
let old_mtu = mmp.path_mtu.current_mtu();
let changed = mmp
.path_mtu
.apply_notification(notif.path_mtu, crate::mmp::mono_ms());
.apply_notification(notif.path_mtu, crate::time::mono_ms());
let new_mtu = mmp.path_mtu.current_mtu();
if !changed {
@@ -1521,7 +1521,7 @@ impl Node {
let old_mtu = mmp.path_mtu.current_mtu();
if mmp
.path_mtu
.apply_notification(msg.mtu, crate::mmp::mono_ms())
.apply_notification(msg.mtu, crate::time::mono_ms())
{
let new_mtu = mmp.path_mtu.current_mtu();
info!(
src/node/session.rs
+1 -1
View File
@@ -668,7 +668,7 @@ impl SessionEntry {
self.rekey_jitter_secs = draw_rekey_jitter();
// Reset MMP counters to avoid metric discontinuity
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
if let Some(mmp) = &mut self.mmp {
mmp.reset_for_rekey(now_ms);
}
src/node/tests/mmp_chartests.rs
+12 -12
View File
@@ -12,12 +12,12 @@
//!
//! Assertions capture what the code does today, surprising or not.
//!
//! Report-generation is probed through the reused `src/mmp/` primitives
//! Report-generation is probed through the reused `proto/mmp/` primitives
//! (`should_send_report` / `should_send_notification`): after a handler tick,
//! a *consumed* interval reads as "not due" (the report was built) while an
//! *ungated* interval still reads as "due" (the report was suppressed by the
//! mode/flag gate). This survives the later refactor because those primitives
//! stay in `src/mmp/` unchanged.
//! stay in `proto/mmp/` unchanged.
//!
//! Two `#[cfg(test)]` production seams are used, both on `ActivePeer`:
//! * `test_init_mmp(mode)` — attach link MMP with a chosen mode to a
@@ -55,7 +55,7 @@ fn arm_link_mmp(node: &mut Node, addr: &NodeAddr) {
let mmp = node.get_peer_mut(addr).unwrap().mmp_mut().unwrap();
mmp.sender.record_sent(1, 100, 500);
mmp.receiver
.record_recv(1, 100, 500, false, crate::mmp::mono_ms());
.record_recv(1, 100, 500, false, crate::time::mono_ms());
}
/// Complete an in-memory Noise IK handshake, returning the initiator session.
@@ -109,7 +109,7 @@ fn arm_session_mmp(node: &mut Node, addr: &NodeAddr) {
let mmp = node.sessions.get_mut(addr).unwrap().mmp_mut().unwrap();
mmp.sender.record_sent(1, 100, 500);
mmp.receiver
.record_recv(1, 100, 500, false, crate::mmp::mono_ms());
.record_recv(1, 100, 500, false, crate::time::mono_ms());
}
// ===========================================================================
@@ -127,7 +127,7 @@ async fn mmp_full_mode_builds_sender_and_receiver_reports() {
node.check_mmp_reports().await;
let mmp = node.get_peer(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
let now = crate::time::mono_ms();
assert!(
!mmp.sender.should_send_report(now),
"Full mode consumes the sender interval (SenderReport built)"
@@ -149,7 +149,7 @@ async fn mmp_lightweight_mode_builds_receiver_report_only() {
node.check_mmp_reports().await;
let mmp = node.get_peer(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
let now = crate::time::mono_ms();
assert!(
mmp.sender.should_send_report(now),
"Lightweight mode suppresses the SenderReport (sender interval intact)"
@@ -170,7 +170,7 @@ async fn mmp_minimal_mode_builds_nothing() {
node.check_mmp_reports().await;
let mmp = node.get_peer(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
let now = crate::time::mono_ms();
assert!(
mmp.sender.should_send_report(now),
"Minimal mode suppresses the SenderReport"
@@ -194,7 +194,7 @@ async fn mmp_should_log_marks_logged_once_per_interval() {
.unwrap()
.mmp()
.unwrap()
.should_log(crate::mmp::mono_ms()),
.should_log(crate::time::mono_ms()),
"a freshly created peer is due for its first operator log"
);
@@ -206,7 +206,7 @@ async fn mmp_should_log_marks_logged_once_per_interval() {
.unwrap()
.mmp()
.unwrap()
.should_log(crate::mmp::mono_ms()),
.should_log(crate::time::mono_ms()),
"after one tick the log is marked and not due again within the interval"
);
}
@@ -226,7 +226,7 @@ async fn session_full_mode_builds_sender_and_receiver_reports() {
node.check_session_mmp_reports().await;
let mmp = node.get_session(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
let now = crate::time::mono_ms();
assert!(
!mmp.sender.should_send_report(now),
"Full session consumes the sender interval"
@@ -254,7 +254,7 @@ async fn session_minimal_mode_still_sends_path_mtu() {
.path_mtu
.observe_incoming_mtu(1200);
let now_before = crate::mmp::mono_ms();
let now_before = crate::time::mono_ms();
assert!(
node.get_session(&addr)
.unwrap()
@@ -268,7 +268,7 @@ async fn session_minimal_mode_still_sends_path_mtu() {
node.check_session_mmp_reports().await;
let mmp = node.get_session(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
let now = crate::time::mono_ms();
assert!(
mmp.sender.should_send_report(now),
"Minimal mode suppresses the session SenderReport"
src/node/tests/session.rs
+1 -1
View File
@@ -1912,7 +1912,7 @@ async fn test_tun_outbound_path_mtu_generates_ptb() {
let entry = nodes[0].node.get_session_mut(&node1_addr).unwrap();
let mmp = entry.mmp_mut().unwrap();
mmp.path_mtu
.apply_notification(reduced_mtu, crate::mmp::mono_ms());
.apply_notification(reduced_mtu, crate::time::mono_ms());
assert_eq!(mmp.path_mtu.current_mtu(), reduced_mtu);
}
src/node/tree.rs
+3 -3
View File
@@ -328,7 +328,7 @@ impl Node {
// Monotonic ms for the flap-dampening / hold-down timers (distinct from
// the wall-clock `now_ms` above used for the peer's tree position). Read
// once and threaded into classify + the state mutators.
let mono_now_ms = crate::mmp::mono_ms();
let mono_now_ms = crate::time::mono_ms();
match Stp::classify_announce(&self.tree_state, *from, &peer_costs, &skip, mono_now_ms) {
TreeDecision::Switch {
@@ -582,7 +582,7 @@ impl Node {
// Monotonic ms for the flap-dampening / hold-down timers, read once and
// threaded into classify + the state mutators.
let mono_now_ms = crate::mmp::mono_ms();
let mono_now_ms = crate::time::mono_ms();
match Stp::classify_periodic(&self.tree_state, &peer_costs, &skip, mono_now_ms) {
TreeDecision::Switch {
@@ -735,7 +735,7 @@ impl Node {
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let mono_now_ms = crate::mmp::mono_ms();
let mono_now_ms = crate::time::mono_ms();
// Removal is not a pure classify: `handle_parent_lost` is a &mut mutator
// whose returned `changed` bool IS the decision. Drive it and map the
src/peer/active.rs
+3 -3
View File
@@ -3,7 +3,7 @@
//! Represents a fully authenticated peer after successful Noise handshake.
//! ActivePeer holds tree state, Bloom filter, and routing information.
use crate::mmp::MmpConfig;
use crate::config::MmpConfig;
use crate::node::REKEY_JITTER_SECS;
use crate::noise::{HandshakeState as NoiseHandshakeState, NoiseError, NoiseSession};
use crate::proto::bloom::BloomFilter;
@@ -1056,7 +1056,7 @@ impl ActivePeer {
self.reset_replay_suppressed();
// Reset MMP counters to avoid metric discontinuity
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
if let Some(mmp) = &mut self.mmp {
mmp.reset_for_rekey(now_ms);
}
@@ -1093,7 +1093,7 @@ impl ActivePeer {
self.reset_replay_suppressed();
// Reset MMP counters to avoid metric discontinuity
let now_ms = crate::mmp::mono_ms();
let now_ms = crate::time::mono_ms();
if let Some(mmp) = &mut self.mmp {
mmp.reset_for_rekey(now_ms);
}
src/proto/mmp/core.rs
+5 -5
View File
@@ -6,7 +6,7 @@
//! (pre-computing every clock read into plain `bool` snapshot fields, and
//! resolving the rekey-suppression predicate shell-side), call the `plan_*`
//! decisions, and drive the returned [`MmpAction`]s — the actual sends,
//! registry mutations, `src/mmp/` primitive calls, metrics, and logging. No
//! registry mutations, `proto/mmp/` primitive calls, metrics, and logging. No
//! I/O, no clock, no metrics, no logging here.
//!
//! Unlike routing/discovery (whose cores do live cross-table reads through a
@@ -50,7 +50,7 @@ pub(crate) struct PeerLivenessSnapshot {
///
/// Every timing read is resolved shell-side into a plain `bool`: `sr_due` /
/// `rr_due` are the pre-evaluated `should_send_report` gates on the peer's
/// `src/mmp/` sender/receiver primitives, and `log_due` is the pre-evaluated
/// `proto/mmp/` sender/receiver primitives, and `log_due` is the pre-evaluated
/// `should_log` gate. `send_sr`/`send_rr` are the profile provide/want flags:
/// on the master (IK) line there is no profile negotiation, so the shell sets
/// them to the behavior-neutral constant `true` (ANDing a runtime `true` is a
@@ -81,7 +81,7 @@ pub(crate) struct LinkReportSnapshot {
///
/// Every timing read is resolved shell-side into a plain `bool`: `sr_due` /
/// `rr_due` are the pre-evaluated `should_send_report` gates on the session's
/// `src/mmp/` sender/receiver primitives, `mtu_due` is the pre-evaluated
/// `proto/mmp/` sender/receiver primitives, `mtu_due` is the pre-evaluated
/// `path_mtu.should_send_notification` gate, and `log_due` is the pre-evaluated
/// `should_log` gate. Unlike the link path there is **no** `send_sr`/`send_rr`
/// profile flag — the session handler has never gated on a profile, so no
@@ -172,7 +172,7 @@ pub(crate) enum MmpAction {
/// Send a heartbeat to `peer`: the shell runs `mark_heartbeat_sent` and the
/// encrypted link send.
Heartbeat { peer: NodeAddr },
/// Build (shell: `src/mmp/` `build_report` + `encode`) and send the given
/// Build (shell: `proto/mmp/` `build_report` + `encode`) and send the given
/// link report over the encrypted link. The interval-advancing
/// `build_report` mutation happens **only** while driving this action, so an
/// ungated report never advances its interval.
@@ -183,7 +183,7 @@ pub(crate) enum MmpAction {
/// Emit the periodic link operator log for `peer` (shell owns the `tracing`
/// call and runs `mark_logged`).
LogLink { peer: NodeAddr },
/// Build (shell: `src/mmp/` `build_report`/`build_notification` + the
/// Build (shell: `proto/mmp/` `build_report`/`build_notification` + the
/// `Session*`/`PathMtuNotification` codec) and send the given session report
/// over the encrypted session. The interval/notification-advancing build
/// mutation happens **only** while driving this action.
src/proto/mmp/limits.rs
+55
View File
@@ -0,0 +1,55 @@
//! MMP tuning constants (EWMA parameters, report-interval bounds, defaults).
//!
//! The plain values the owned sender/receiver/metrics state machines and the
//! shell config clamp against. Re-exported from the module root so external
//! callers keep the `crate::proto::mmp::<CONST>` path.
// --- EWMA parameters ---
/// Dual EWMA short-term: α = 1/4.
pub const EWMA_SHORT_ALPHA: f64 = 0.25;
/// Dual EWMA long-term: α = 1/32.
pub const EWMA_LONG_ALPHA: f64 = 1.0 / 32.0;
// --- Timing defaults (milliseconds) ---
/// Default report interval before SRTT is available (cold start).
pub const DEFAULT_COLD_START_INTERVAL_MS: u64 = 200;
/// Minimum report interval (SRTT clamp floor).
///
/// Raised from 100ms to 1000ms: parent re-evaluation runs every 60s,
/// so 60 samples/cycle is more than sufficient for EWMA convergence (~10).
/// The cold-start phase uses `DEFAULT_COLD_START_INTERVAL_MS` (200ms) for
/// fast initial SRTT convergence before transitioning to this floor.
pub const MIN_REPORT_INTERVAL_MS: u64 = 1_000;
/// Maximum report interval (SRTT clamp ceiling).
pub const MAX_REPORT_INTERVAL_MS: u64 = 5_000;
/// Number of SRTT samples before transitioning from cold-start to normal floor.
///
/// During cold-start, report intervals use `DEFAULT_COLD_START_INTERVAL_MS` as
/// the floor to gather SRTT samples quickly. After this many updates, the floor
/// switches to `MIN_REPORT_INTERVAL_MS`.
pub const COLD_START_SAMPLES: u32 = 5;
/// Default OWD ring buffer capacity.
pub const DEFAULT_OWD_WINDOW_SIZE: usize = 32;
/// Default operator log interval in seconds.
pub const DEFAULT_LOG_INTERVAL_SECS: u64 = 30;
// --- Session-layer timing defaults ---
// Session reports are routed end-to-end (bandwidth cost on every transit link),
// so intervals are higher than link-layer.
/// Session-layer minimum report interval.
pub const MIN_SESSION_REPORT_INTERVAL_MS: u64 = 500;
/// Session-layer maximum report interval.
pub const MAX_SESSION_REPORT_INTERVAL_MS: u64 = 10_000;
/// Session-layer cold-start report interval (before SRTT is available).
pub const SESSION_COLD_START_INTERVAL_MS: u64 = 1_000;
src/proto/mmp/metrics.rs
+333
View File
@@ -0,0 +1,333 @@
//! Derived MMP metrics (sans-IO).
//!
//! Sender-side derived metrics (RTT, loss, goodput, ETX, trends) computed from
//! incoming `ReceiverReport`s, plus the [`RrLog`] observability outcome returned
//! to the shell so the operator `trace!` points can fire there rather than
//! mid-decision. All time inputs are injected `u64` milliseconds.
use super::algorithms::{DualEwma, SrttEstimator, compute_etx};
use super::wire::ReceiverReport;
// ============================================================================
// ReceiverReport processing outcome (observability hoisted to the shell)
// ============================================================================
/// The observability outcome of [`MmpMetrics::process_receiver_report`], returned
/// so the async shell can emit the operator `trace!` points that used to live
/// mid-decision (per the sans-IO rule that migrated code carries no `tracing`).
///
/// Exactly one variant applies per call; the shell has the incoming report and
/// our timestamp, so this only carries the values the shell cannot otherwise
/// reconstruct (the previous cumulative snapshot / the derived RTT).
pub enum RrLog {
/// No RTT log point fires (the report carried no timestamp echo).
None,
/// The report was ignored as stale/duplicate. Carries the previous
/// cumulative snapshot for the operator trace.
Stale {
prev_highest: u64,
prev_packets: u64,
prev_bytes: u64,
},
/// A valid RTT sample was taken; carries the derived RTT and the SRTT as it
/// stood **before** this sample was folded in (matching the original log,
/// which read `srtt` before `update`).
RttSample { rtt_ms: u32, srtt_ms: f64 },
/// The report carried an echo but the derived RTT was invalid (wrapped /
/// non-positive).
InvalidRtt,
}
// ============================================================================
// MmpMetrics (derived metrics)
// ============================================================================
/// Derived MMP metrics, updated from incoming ReceiverReports.
///
/// This lives on the sender side: when we receive a ReceiverReport from
/// our peer describing what they observed about our traffic, we process
/// it here to compute RTT, loss, goodput, and trend indicators.
pub struct MmpMetrics {
/// Smoothed RTT from timestamp echo.
pub srtt: SrttEstimator,
/// Dual EWMA trend detectors.
pub rtt_trend: DualEwma,
pub loss_trend: DualEwma,
pub goodput_trend: DualEwma,
pub jitter_trend: DualEwma,
pub etx_trend: DualEwma,
/// Forward delivery ratio (what fraction of our frames the peer received).
pub delivery_ratio_forward: f64,
/// Reverse delivery ratio (set when we compute from our own receiver state).
pub delivery_ratio_reverse: f64,
/// ETX computed from bidirectional delivery ratios.
pub etx: f64,
/// Smoothed goodput in bytes/sec (forward direction: what the peer received from us).
pub goodput_bps: f64,
// --- State for delta computation ---
/// Previous ReceiverReport's cumulative counters (for computing interval deltas).
prev_rr_cum_packets: u64,
prev_rr_cum_bytes: u64,
prev_rr_highest_counter: u64,
prev_rr_ecn_ce: u32,
prev_rr_reorder: u32,
/// Time (injected `u64` ms) of previous ReceiverReport (for goodput rate).
prev_rr_ms: Option<u64>,
/// Whether we have a previous ReceiverReport for delta computation.
has_prev_rr: bool,
// --- State for reverse delivery ratio delta computation ---
/// Previous reverse-side cumulative packets received (our receiver state).
prev_reverse_packets: u64,
/// Previous reverse-side highest counter (our receiver state).
prev_reverse_highest: u64,
/// Whether we have a previous reverse-side snapshot for delta computation.
has_prev_reverse: bool,
}
impl MmpMetrics {
/// Reset state derived from ReceiverReport counters for rekey cutover.
///
/// The new session starts with counter 0, so the prev_rr deltas must
/// be reset to avoid computing bogus loss/goodput from the counter
/// discontinuity. RTT (SRTT) is preserved since it remains valid.
pub fn reset_for_rekey(&mut self) {
self.prev_rr_cum_packets = 0;
self.prev_rr_cum_bytes = 0;
self.prev_rr_highest_counter = 0;
self.prev_rr_ecn_ce = 0;
self.prev_rr_reorder = 0;
self.prev_rr_ms = None;
self.has_prev_rr = false;
self.delivery_ratio_forward = 1.0;
self.prev_reverse_packets = 0;
self.prev_reverse_highest = 0;
self.has_prev_reverse = false;
// Keep srtt, etx, trends, goodput_bps — they'll refresh from data
}
pub fn new() -> Self {
Self {
srtt: SrttEstimator::new(),
rtt_trend: DualEwma::new(),
loss_trend: DualEwma::new(),
goodput_trend: DualEwma::new(),
jitter_trend: DualEwma::new(),
etx_trend: DualEwma::new(),
delivery_ratio_forward: 1.0,
delivery_ratio_reverse: 1.0,
etx: 1.0,
goodput_bps: 0.0,
prev_rr_cum_packets: 0,
prev_rr_cum_bytes: 0,
prev_rr_highest_counter: 0,
prev_rr_ecn_ce: 0,
prev_rr_reorder: 0,
prev_rr_ms: None,
has_prev_rr: false,
prev_reverse_packets: 0,
prev_reverse_highest: 0,
has_prev_reverse: false,
}
}
/// Process an incoming ReceiverReport (from the peer about our traffic).
///
/// `our_timestamp_ms` is the current session-relative time in ms (for RTT).
/// `now_ms` is the injected monotonic time in ms (for goodput rate).
///
/// Returns `(first_srtt, log)`: `first_srtt` is `true` if this report
/// produced the first SRTT measurement (transition from uninitialized to
/// initialized); `log` is the [`RrLog`] observability outcome the shell
/// emits (the `tracing` that used to fire here now lives shell-side).
pub fn process_receiver_report(
&mut self,
rr: &ReceiverReport,
our_timestamp_ms: u32,
now_ms: u64,
) -> (bool, RrLog) {
let had_srtt = self.srtt.initialized();
if self.has_prev_rr {
let counters_regressed = rr.highest_counter < self.prev_rr_highest_counter
|| rr.cumulative_packets_recv < self.prev_rr_cum_packets
|| rr.cumulative_bytes_recv < self.prev_rr_cum_bytes
|| rr.ecn_ce_count < self.prev_rr_ecn_ce
|| rr.cumulative_reorder_count < self.prev_rr_reorder;
let duplicate_counters = rr.highest_counter == self.prev_rr_highest_counter
&& rr.cumulative_packets_recv == self.prev_rr_cum_packets
&& rr.cumulative_bytes_recv == self.prev_rr_cum_bytes
&& rr.ecn_ce_count == self.prev_rr_ecn_ce
&& rr.cumulative_reorder_count == self.prev_rr_reorder;
// Safe to drop: reports are only built after interval data, so
// a fresh report always advances at least one cumulative counter.
if counters_regressed || duplicate_counters {
return (
false,
RrLog::Stale {
prev_highest: self.prev_rr_highest_counter,
prev_packets: self.prev_rr_cum_packets,
prev_bytes: self.prev_rr_cum_bytes,
},
);
}
}
let mut log = RrLog::None;
// --- RTT from timestamp echo ---
// RTT = now - echoed_timestamp - dwell_time
if rr.timestamp_echo > 0 {
let echo_ms = rr.timestamp_echo;
let dwell_ms = u32::from(rr.dwell_time);
let rtt_sample_ms = echo_ms
.checked_add(dwell_ms)
.and_then(|send_done_ms| our_timestamp_ms.checked_sub(send_done_ms));
match rtt_sample_ms {
Some(rtt_ms) if rtt_ms > 0 => {
let rtt_us = (rtt_ms as i64) * 1000;
// Capture SRTT before folding in this sample (the original
// trace read `srtt` before `update`).
let srtt_ms = self.srtt.srtt_us() as f64 / 1000.0;
log = RrLog::RttSample { rtt_ms, srtt_ms };
self.srtt.update(rtt_us);
self.rtt_trend.update(rtt_us as f64);
}
_ => {
log = RrLog::InvalidRtt;
}
}
}
// --- Loss rate from cumulative counters ---
// Delta: frames the peer should have received vs. actually received
if self.has_prev_rr {
let counter_span = rr
.highest_counter
.saturating_sub(self.prev_rr_highest_counter);
let packets_delta = rr
.cumulative_packets_recv
.saturating_sub(self.prev_rr_cum_packets);
if counter_span > 0 {
let delivery = (packets_delta as f64) / (counter_span as f64);
self.delivery_ratio_forward = delivery.clamp(0.0, 1.0);
let loss_rate = 1.0 - self.delivery_ratio_forward;
self.loss_trend.update(loss_rate);
self.etx = compute_etx(self.delivery_ratio_forward, self.delivery_ratio_reverse);
self.etx_trend.update(self.etx);
}
}
// --- Goodput from cumulative bytes + time delta ---
if self.has_prev_rr {
let bytes_delta = rr
.cumulative_bytes_recv
.saturating_sub(self.prev_rr_cum_bytes);
self.goodput_trend.update(bytes_delta as f64);
// Compute bytes/sec if we have a time reference
if let Some(prev_ms) = self.prev_rr_ms {
let secs = (now_ms.saturating_sub(prev_ms) as f64) / 1000.0;
if secs > 0.0 {
let bps = bytes_delta as f64 / secs;
// EWMA smoothing: α = 1/4
if self.goodput_bps == 0.0 {
self.goodput_bps = bps;
} else {
self.goodput_bps += (bps - self.goodput_bps) * 0.25;
}
}
}
}
// --- Jitter trend ---
self.jitter_trend.update(rr.jitter as f64);
// --- Save for next delta ---
self.prev_rr_cum_packets = rr.cumulative_packets_recv;
self.prev_rr_cum_bytes = rr.cumulative_bytes_recv;
self.prev_rr_highest_counter = rr.highest_counter;
self.prev_rr_ecn_ce = rr.ecn_ce_count;
self.prev_rr_reorder = rr.cumulative_reorder_count;
self.prev_rr_ms = Some(now_ms);
self.has_prev_rr = true;
(!had_srtt && self.srtt.initialized(), log)
}
/// Update the reverse delivery ratio from our own receiver state.
///
/// Computes a per-interval delta (same as forward ratio) rather than
/// a lifetime cumulative ratio, so ETX responds to recent conditions.
pub fn update_reverse_delivery(&mut self, our_recv_packets: u64, peer_highest: u64) {
if self.has_prev_reverse {
let counter_span = peer_highest.saturating_sub(self.prev_reverse_highest);
let packets_delta = our_recv_packets.saturating_sub(self.prev_reverse_packets);
if counter_span > 0 {
let delivery = (packets_delta as f64) / (counter_span as f64);
self.delivery_ratio_reverse = delivery.clamp(0.0, 1.0);
self.etx = compute_etx(self.delivery_ratio_forward, self.delivery_ratio_reverse);
self.etx_trend.update(self.etx);
}
}
self.prev_reverse_packets = our_recv_packets;
self.prev_reverse_highest = peer_highest;
self.has_prev_reverse = true;
}
/// Current smoothed RTT in milliseconds, or `None` if not yet measured.
pub fn srtt_ms(&self) -> Option<f64> {
if self.srtt.initialized() {
Some(self.srtt.srtt_us() as f64 / 1000.0)
} else {
None
}
}
/// Current loss rate (0.0 = no loss, 1.0 = total loss).
pub fn loss_rate(&self) -> f64 {
1.0 - self.delivery_ratio_forward
}
/// Smoothed loss rate (long-term EWMA), or `None` if not yet initialized.
pub fn smoothed_loss(&self) -> Option<f64> {
if self.loss_trend.initialized() {
Some(self.loss_trend.long())
} else {
None
}
}
/// Smoothed ETX (long-term EWMA), or `None` if not yet initialized.
pub fn smoothed_etx(&self) -> Option<f64> {
if self.etx_trend.initialized() {
Some(self.etx_trend.long())
} else {
None
}
}
/// Current smoothed goodput in bytes/sec, or 0 if not yet measured.
pub fn goodput_bps(&self) -> f64 {
self.goodput_bps
}
/// Cumulative ECN CE count from the most recent ReceiverReport.
pub fn last_ecn_ce_count(&self) -> u32 {
self.prev_rr_ecn_ce
}
}
impl Default for MmpMetrics {
fn default() -> Self {
Self::new()
}
}
src/proto/mmp/mod.rs
+21 -52
View File
@@ -7,8 +7,12 @@
//! - `core.rs` — the snapshot read-seams, the [`MmpAction`] effect vocabulary,
//! and the pure `plan_*` decisions (line-invariant across the master/next
//! report shapes).
//! - `state.rs` — [`Mmp`] (the reporting anchor) plus the owned sender/receiver/
//! metrics/path-MTU state machines (`u64`-ms time seam, `no_std`+`alloc`).
//! - `state.rs` — [`Mmp`] (the reporting anchor) plus the [`MmpPeerState`]/
//! [`MmpSessionState`] per-entity aggregates.
//! - `sender.rs`/`receiver.rs`/`metrics.rs`/`path_mtu.rs` — the owned sender/
//! receiver/derived-metrics/path-MTU role state machines (`u64`-ms time seam,
//! `no_std`+`alloc`).
//! - `limits.rs` — the module tuning constants.
//! - `algorithms.rs` — the pure estimators (jitter/SRTT/dual-EWMA/OWD-trend/ETX)
//! and the spin-bit state (`no_std`+`alloc`).
//! - `wire.rs` — the link-layer and session-layer report codecs.
@@ -23,6 +27,11 @@ use serde::{Deserialize, Serialize};
mod algorithms;
mod core;
mod limits;
mod metrics;
mod path_mtu;
mod receiver;
mod sender;
mod state;
mod wire;
@@ -34,7 +43,8 @@ pub(crate) use core::{
BackoffUpdate, LinkReportKind, LinkReportSnapshot, MmpAction, PeerLivenessSnapshot, SendResult,
SessionReportKind, SessionReportSnapshot,
};
pub(crate) use state::{Mmp, MmpMetrics, MmpPeerState, MmpSessionState, RrLog};
pub(crate) use metrics::{MmpMetrics, RrLog};
pub(crate) use state::{Mmp, MmpPeerState, MmpSessionState};
pub use wire::{
PathMtuNotification, ReceiverReport, SenderReport, SessionReceiverReport, SessionSenderReport,
};
@@ -70,52 +80,11 @@ impl fmt::Display for MmpMode {
// Constants
// ============================================================================
// --- EWMA parameters ---
/// Dual EWMA short-term: α = 1/4.
pub const EWMA_SHORT_ALPHA: f64 = 0.25;
/// Dual EWMA long-term: α = 1/32.
pub const EWMA_LONG_ALPHA: f64 = 1.0 / 32.0;
// --- Timing defaults (milliseconds) ---
/// Default report interval before SRTT is available (cold start).
pub const DEFAULT_COLD_START_INTERVAL_MS: u64 = 200;
/// Minimum report interval (SRTT clamp floor).
///
/// Raised from 100ms to 1000ms: parent re-evaluation runs every 60s,
/// so 60 samples/cycle is more than sufficient for EWMA convergence (~10).
/// The cold-start phase uses `DEFAULT_COLD_START_INTERVAL_MS` (200ms) for
/// fast initial SRTT convergence before transitioning to this floor.
pub const MIN_REPORT_INTERVAL_MS: u64 = 1_000;
/// Maximum report interval (SRTT clamp ceiling).
pub const MAX_REPORT_INTERVAL_MS: u64 = 5_000;
/// Number of SRTT samples before transitioning from cold-start to normal floor.
///
/// During cold-start, report intervals use `DEFAULT_COLD_START_INTERVAL_MS` as
/// the floor to gather SRTT samples quickly. After this many updates, the floor
/// switches to `MIN_REPORT_INTERVAL_MS`.
pub const COLD_START_SAMPLES: u32 = 5;
/// Default OWD ring buffer capacity.
pub const DEFAULT_OWD_WINDOW_SIZE: usize = 32;
/// Default operator log interval in seconds.
pub const DEFAULT_LOG_INTERVAL_SECS: u64 = 30;
// --- Session-layer timing defaults ---
// Session reports are routed end-to-end (bandwidth cost on every transit link),
// so intervals are higher than link-layer.
/// Session-layer minimum report interval.
pub const MIN_SESSION_REPORT_INTERVAL_MS: u64 = 500;
/// Session-layer maximum report interval.
pub const MAX_SESSION_REPORT_INTERVAL_MS: u64 = 10_000;
/// Session-layer cold-start report interval (before SRTT is available).
pub const SESSION_COLD_START_INTERVAL_MS: u64 = 1_000;
// The module tuning constants live in `limits.rs`; re-exported here so external
// callers keep the `crate::proto::mmp::<CONST>` path.
pub use limits::{
COLD_START_SAMPLES, DEFAULT_COLD_START_INTERVAL_MS, DEFAULT_LOG_INTERVAL_SECS,
DEFAULT_OWD_WINDOW_SIZE, EWMA_LONG_ALPHA, EWMA_SHORT_ALPHA, MAX_REPORT_INTERVAL_MS,
MAX_SESSION_REPORT_INTERVAL_MS, MIN_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS,
SESSION_COLD_START_INTERVAL_MS,
};
src/proto/mmp/path_mtu.rs
+175
View File
@@ -0,0 +1,175 @@
//! Path MTU tracking for a single session (sans-IO, session-layer only).
//!
//! Destination side observes `path_mtu` from incoming envelopes and emits
//! notifications; source side applies received notifications to bound outbound
//! datagram size. All time inputs are injected `u64` milliseconds.
/// Path MTU tracking for a single session.
///
/// Destination side: observes `path_mtu` from incoming SessionDatagram envelopes
/// and generates PathMtuNotification messages back to the source.
///
/// Source side: applies received PathMtuNotification to limit outbound datagram
/// size. Decrease is immediate; increase requires 3 consecutive notifications.
pub struct PathMtuState {
/// Current effective path MTU (what we use for sending).
current_mtu: u16,
/// Last observed path MTU from incoming datagrams (destination-side).
last_observed_mtu: u16,
/// Whether the observed MTU has changed since the last notification.
observed_changed: bool,
/// Last time (injected `u64` ms) a PathMtuNotification was sent.
last_notification_ms: Option<u64>,
/// Notification interval in ms: max(10s, 5 * SRTT). Default 10s.
notification_interval_ms: u64,
/// For source-side increase tracking: consecutive higher-value notifications.
consecutive_increase_count: u8,
/// Time (injected `u64` ms) of the first notification in the current
/// increase sequence.
first_increase_ms: Option<u64>,
/// The MTU value being proposed for increase.
pending_increase_mtu: u16,
}
impl PathMtuState {
/// Create path MTU state with no initial measurement.
pub fn new() -> Self {
Self {
current_mtu: u16::MAX,
last_observed_mtu: u16::MAX,
observed_changed: false,
last_notification_ms: None,
notification_interval_ms: 10_000,
consecutive_increase_count: 0,
first_increase_ms: None,
pending_increase_mtu: 0,
}
}
/// Current effective path MTU (source-side, for sending).
pub fn current_mtu(&self) -> u16 {
self.current_mtu
}
/// Last observed incoming path MTU (destination-side).
pub fn last_observed_mtu(&self) -> u16 {
self.last_observed_mtu
}
/// Update notification interval from SRTT: max(10s, 5 * SRTT).
pub fn update_interval_from_srtt(&mut self, srtt_ms: f64) {
self.notification_interval_ms = ((srtt_ms * 5.0) as u64).max(10_000);
}
/// Seed source-side current_mtu from outbound transport MTU.
///
/// Called on each send. Only decreases (never increases) the current_mtu
/// so the destination's PathMtuNotification can still raise it later.
/// Ensures current_mtu doesn't stay at u16::MAX before any notification
/// arrives from the destination.
pub fn seed_source_mtu(&mut self, outbound_mtu: u16) {
if outbound_mtu < self.current_mtu {
self.current_mtu = outbound_mtu;
}
}
// --- Destination side ---
/// Observe the path_mtu from an incoming SessionDatagram envelope.
///
/// Called on the destination (receiver) side for every session message.
pub fn observe_incoming_mtu(&mut self, path_mtu: u16) {
if path_mtu != self.last_observed_mtu {
self.observed_changed = true;
self.last_observed_mtu = path_mtu;
}
}
/// Check if a PathMtuNotification should be sent.
///
/// Send on first measurement, on decrease (immediate), or periodic
/// confirmation at the notification interval. `now_ms` is the injected
/// monotonic time in milliseconds.
pub fn should_send_notification(&self, now_ms: u64) -> bool {
if self.last_observed_mtu == u16::MAX {
return false; // No measurement yet
}
match self.last_notification_ms {
None => true, // First measurement
Some(last) => {
// Immediate on decrease
if self.observed_changed && self.last_observed_mtu < self.current_mtu {
return true;
}
// Periodic confirmation
now_ms.saturating_sub(last) >= self.notification_interval_ms
}
}
}
/// Build a PathMtuNotification from current state.
///
/// Returns the path_mtu value to send. Caller handles encoding.
pub fn build_notification(&mut self, now_ms: u64) -> Option<u16> {
if self.last_observed_mtu == u16::MAX {
return None;
}
self.last_notification_ms = Some(now_ms);
self.observed_changed = false;
Some(self.last_observed_mtu)
}
// --- Source side ---
/// Apply a received PathMtuNotification.
///
/// - Decrease: immediate (take the lower value).
/// - Increase: require 3 consecutive notifications with the same higher
/// value, spanning at least 2 * notification_interval.
///
/// `now_ms` is the injected monotonic time in milliseconds. Returns `true`
/// if the effective MTU changed.
pub fn apply_notification(&mut self, reported_mtu: u16, now_ms: u64) -> bool {
if reported_mtu < self.current_mtu {
// Decrease: immediate
self.current_mtu = reported_mtu;
self.consecutive_increase_count = 0;
self.first_increase_ms = None;
return true;
}
if reported_mtu > self.current_mtu {
// Increase: track consecutive notifications
if reported_mtu == self.pending_increase_mtu {
self.consecutive_increase_count += 1;
} else {
// Different value: reset sequence
self.pending_increase_mtu = reported_mtu;
self.consecutive_increase_count = 1;
self.first_increase_ms = Some(now_ms);
}
// Accept increase after 3 consecutive spanning 2 * interval
if self.consecutive_increase_count >= 3
&& let Some(first_ms) = self.first_increase_ms
{
let required_ms = self.notification_interval_ms * 2;
if now_ms.saturating_sub(first_ms) >= required_ms {
self.current_mtu = reported_mtu;
self.consecutive_increase_count = 0;
self.first_increase_ms = None;
return true;
}
}
}
// No change (equal or increase not yet confirmed)
false
}
}
impl Default for PathMtuState {
fn default() -> Self {
Self::new()
}
}
src/proto/mmp/receiver.rs
+447
View File
@@ -0,0 +1,447 @@
//! Per-peer receiver-side MMP state (sans-IO).
//!
//! Accumulates per-frame observations (loss bursts, jitter, OWD trend, ECN)
//! and produces `ReceiverReport` snapshots. All time inputs are injected `u64`
//! milliseconds.
use super::algorithms::{JitterEstimator, OwdTrendDetector};
use super::wire::ReceiverReport;
use super::{
COLD_START_SAMPLES, DEFAULT_COLD_START_INTERVAL_MS, DEFAULT_OWD_WINDOW_SIZE,
MAX_REPORT_INTERVAL_MS, MIN_REPORT_INTERVAL_MS,
};
/// Grace period after rekey before resuming jitter calculation.
///
/// During rekey cutover, frames from the old session may still arrive via the
/// drain window (DRAIN_WINDOW_SECS = 10s). These carry large sender timestamps
/// from the old session, producing enormous transit deltas that spike the EWMA
/// jitter estimator. We suppress jitter updates for drain window + 5s margin.
const REKEY_JITTER_GRACE_SECS: u64 = 15;
// ============================================================================
// Gap Tracker (burst loss detection)
// ============================================================================
/// Tracks counter gaps to detect loss bursts.
///
/// Each gap in the counter sequence is a burst of lost frames.
/// Maintains per-interval statistics that are reset when a report is built.
struct GapTracker {
/// Next expected counter value.
expected_next: Option<u64>,
/// Whether we are currently in a burst (gap).
in_burst: bool,
/// Length of the current burst.
current_burst_len: u16,
// --- Per-interval stats (reset on report) ---
/// Number of distinct burst events this interval.
burst_count: u32,
/// Longest burst in this interval.
max_burst_len: u16,
/// Sum of all burst lengths (for mean computation).
total_burst_len: u64,
}
impl GapTracker {
fn new() -> Self {
Self {
expected_next: None,
in_burst: false,
current_burst_len: 0,
burst_count: 0,
max_burst_len: 0,
total_burst_len: 0,
}
}
/// Process a received counter value. Returns the number of lost frames
/// detected (0 if in order or first frame).
fn observe(&mut self, counter: u64) -> u64 {
let Some(expected) = self.expected_next else {
// First frame: initialize
self.expected_next = Some(counter + 1);
return 0;
};
let lost = if counter > expected {
// Gap detected
let gap = counter - expected;
if self.in_burst {
// Extend current burst
self.current_burst_len = self.current_burst_len.saturating_add(gap as u16);
} else {
// New burst
self.in_burst = true;
self.current_burst_len = gap as u16;
self.burst_count += 1;
}
gap
} else {
// In-order or duplicate (counter <= expected)
if self.in_burst {
// End current burst
self.finish_burst();
}
0
};
// Update expected (always advance to counter+1 or keep expected if
// this was a late/reordered frame)
if counter >= expected {
self.expected_next = Some(counter + 1);
}
lost
}
/// Finish the current burst and record its stats.
fn finish_burst(&mut self) {
if self.in_burst {
self.max_burst_len = self.max_burst_len.max(self.current_burst_len);
self.total_burst_len += self.current_burst_len as u64;
self.in_burst = false;
self.current_burst_len = 0;
}
}
/// Get interval stats and reset for next interval.
fn take_interval_stats(&mut self) -> (u32, u16, u16) {
// Finish any in-progress burst
self.finish_burst();
let count = self.burst_count;
let max_len = self.max_burst_len;
let mean_len = if count > 0 {
// u8.8 fixed-point: (total / count) * 256
let mean_f = (self.total_burst_len as f64) / (count as f64);
(mean_f * 256.0) as u16
} else {
0
};
// Reset interval
self.burst_count = 0;
self.max_burst_len = 0;
self.total_burst_len = 0;
(count, max_len, mean_len)
}
}
// ============================================================================
// ReceiverState
// ============================================================================
/// Per-peer receiver-side MMP state.
///
/// Accumulates per-frame observations and produces `ReceiverReport` snapshots.
pub struct ReceiverState {
// --- Cumulative (lifetime) ---
cumulative_packets_recv: u64,
cumulative_bytes_recv: u64,
cumulative_reorder_count: u64,
/// Highest counter value ever received.
highest_counter: u64,
// --- Current interval ---
interval_packets_recv: u32,
interval_bytes_recv: u32,
// --- Jitter ---
jitter: JitterEstimator,
// --- OWD trend ---
owd_trend: OwdTrendDetector,
/// Monotonic sequence counter for OWD samples.
owd_seq: u32,
// --- Loss tracking ---
gap_tracker: GapTracker,
// --- ECN ---
ecn_ce_count: u32,
// --- Timestamp echo ---
/// Sender timestamp from the most recent frame (for echo).
last_sender_timestamp: u32,
/// Local time (injected `u64` ms) when the most recent frame was received
/// (for dwell / jitter computation).
last_recv_ms: Option<u64>,
// --- Rekey grace ---
/// When set, jitter updates are suppressed until this injected-ms instant
/// passes. Prevents drain-window frames from spiking the jitter estimator.
rekey_jitter_grace_until_ms: Option<u64>,
// --- Report timing (injected `u64` ms) ---
last_report_ms: Option<u64>,
report_interval_ms: u64,
/// Whether any frames have been received since the last report.
interval_has_data: bool,
// --- Cold-start tracking ---
/// Number of SRTT-based interval updates received.
srtt_sample_count: u32,
}
impl ReceiverState {
pub fn new(owd_window_size: usize) -> Self {
Self::new_with_cold_start(owd_window_size, DEFAULT_COLD_START_INTERVAL_MS)
}
/// Create with a custom cold-start interval (ms).
///
/// Used by session-layer MMP which needs a longer initial interval
/// since reports consume bandwidth on every transit link.
pub fn new_with_cold_start(owd_window_size: usize, cold_start_ms: u64) -> Self {
Self {
cumulative_packets_recv: 0,
cumulative_bytes_recv: 0,
cumulative_reorder_count: 0,
highest_counter: 0,
interval_packets_recv: 0,
interval_bytes_recv: 0,
jitter: JitterEstimator::new(),
owd_trend: OwdTrendDetector::new(owd_window_size),
owd_seq: 0,
gap_tracker: GapTracker::new(),
ecn_ce_count: 0,
last_sender_timestamp: 0,
last_recv_ms: None,
rekey_jitter_grace_until_ms: None,
last_report_ms: None,
report_interval_ms: cold_start_ms,
interval_has_data: false,
srtt_sample_count: 0,
}
}
/// Reset counter-dependent state for rekey cutover.
///
/// After cutover, the new session starts with counter 0 and reset
/// timestamps. Without resetting, the old `highest_counter` and
/// `GapTracker.expected_next` cause false reorder/loss detection.
/// `now_ms` is the injected monotonic time in milliseconds.
pub fn reset_for_rekey(&mut self, now_ms: u64) {
self.highest_counter = 0;
self.cumulative_reorder_count = 0;
self.gap_tracker = GapTracker::new();
self.interval_packets_recv = 0;
self.interval_bytes_recv = 0;
self.jitter = JitterEstimator::new();
self.owd_trend.clear();
self.owd_seq = 0;
self.last_sender_timestamp = 0;
self.last_recv_ms = None;
self.rekey_jitter_grace_until_ms = Some(now_ms + REKEY_JITTER_GRACE_SECS * 1000);
self.ecn_ce_count = 0;
self.interval_has_data = false;
// Keep cumulative_packets_recv, cumulative_bytes_recv (lifetime stats)
// Keep last_report_ms, report_interval_ms (report scheduling)
}
/// Record a received frame from this peer.
///
/// Called on the RX path after AEAD decryption, before message dispatch.
///
/// - `counter`: AEAD counter from outer header
/// - `sender_timestamp_ms`: session-relative timestamp from inner header (ms)
/// - `bytes`: wire payload size
/// - `ce_flag`: CE bit from flags byte
/// - `now_ms`: injected monotonic local time in milliseconds
pub fn record_recv(
&mut self,
counter: u64,
sender_timestamp_ms: u32,
bytes: usize,
ce_flag: bool,
now_ms: u64,
) {
self.interval_has_data = true;
self.cumulative_packets_recv += 1;
self.cumulative_bytes_recv += bytes as u64;
self.interval_packets_recv = self.interval_packets_recv.saturating_add(1);
self.interval_bytes_recv = self.interval_bytes_recv.saturating_add(bytes as u32);
// Reordering detection: counter < highest means out-of-order
if counter < self.highest_counter {
self.cumulative_reorder_count += 1;
} else {
self.highest_counter = counter;
}
// Loss/burst detection
let _lost = self.gap_tracker.observe(counter);
// ECN
if ce_flag {
self.ecn_ce_count = self.ecn_ce_count.saturating_add(1);
}
// Jitter: compute transit time delta
// Transit = recv_local - sender_timestamp (in µs for precision)
// We use the injected monotonic ms clock for the local reference.
let sender_us = (sender_timestamp_ms as i64) * 1000;
// We compute the delta between consecutive transits using relative
// millisecond differences (scaled to µs to match the estimator input).
// Skip during post-rekey grace period to avoid drain-window spikes.
let in_grace = self
.rekey_jitter_grace_until_ms
.is_some_and(|deadline| now_ms < deadline);
if !in_grace {
self.rekey_jitter_grace_until_ms = None; // clear expired grace
if let Some(prev_recv) = self.last_recv_ms {
let recv_delta_us = (now_ms.saturating_sub(prev_recv) as i64) * 1000;
let send_delta_us = sender_us - (self.last_sender_timestamp as i64 * 1000);
let transit_delta = (recv_delta_us - send_delta_us) as i32;
self.jitter.update(transit_delta);
}
}
// OWD trend: use sender timestamp as a proxy for send time
// and the injected ms delta from a fixed reference as receive time.
// Since we only need the *trend* (slope), absolute offsets cancel out.
if let Some(first_recv) = self.last_recv_ms.or(Some(now_ms)) {
let recv_offset_us = (now_ms.saturating_sub(first_recv) as i64) * 1000;
let owd_us = recv_offset_us - sender_us;
self.owd_seq = self.owd_seq.wrapping_add(1);
self.owd_trend.push(self.owd_seq, owd_us);
}
// Timestamp echo state
self.last_sender_timestamp = sender_timestamp_ms;
self.last_recv_ms = Some(now_ms);
}
/// Build a ReceiverReport from current state and reset the interval.
///
/// Returns `None` if no frames have been received since the last report.
/// `now_ms` is the injected monotonic time in milliseconds.
pub fn build_report(&mut self, now_ms: u64) -> Option<ReceiverReport> {
if !self.interval_has_data {
return None;
}
// Dwell time: ms between last frame reception and report generation.
// If it no longer fits on the wire, the timestamp echo cannot produce
// a valid RTT sample. Preserve the counters but suppress the echo.
let (timestamp_echo, dwell_time) = self
.last_recv_ms
.map(|t| {
let dwell_ms = now_ms.saturating_sub(t);
if dwell_ms > u64::from(u16::MAX) {
(0, u16::MAX)
} else {
(self.last_sender_timestamp, dwell_ms as u16)
}
})
.unwrap_or((0, 0));
let (burst_count, max_burst, mean_burst) = self.gap_tracker.take_interval_stats();
let report = ReceiverReport {
highest_counter: self.highest_counter,
cumulative_packets_recv: self.cumulative_packets_recv,
cumulative_bytes_recv: self.cumulative_bytes_recv,
timestamp_echo,
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
self.interval_packets_recv = 0;
self.interval_bytes_recv = 0;
self.interval_has_data = false;
self.last_report_ms = Some(now_ms);
Some(report)
}
/// Check if it's time to send a report.
pub fn should_send_report(&self, now_ms: u64) -> bool {
if !self.interval_has_data {
return false;
}
match self.last_report_ms {
None => true,
Some(last) => now_ms.saturating_sub(last) >= self.report_interval_ms,
}
}
/// Update the report interval based on SRTT (link-layer defaults).
///
/// Receiver reports at 1× SRTT clamped to [floor, MAX]. During cold-start
/// (first `COLD_START_SAMPLES` updates), the floor is the cold-start
/// interval (200ms) for fast SRTT convergence. After that, it rises to
/// `MIN_REPORT_INTERVAL_MS` (1000ms) for steady-state efficiency.
pub fn update_report_interval_from_srtt(&mut self, srtt_us: i64) {
self.srtt_sample_count = self.srtt_sample_count.saturating_add(1);
let floor = if self.srtt_sample_count <= COLD_START_SAMPLES {
DEFAULT_COLD_START_INTERVAL_MS
} else {
MIN_REPORT_INTERVAL_MS
};
self.update_report_interval_with_bounds(srtt_us, floor, MAX_REPORT_INTERVAL_MS);
}
/// Update the report interval based on SRTT with custom bounds.
///
/// Used by session-layer MMP which needs higher clamp values since
/// each report consumes bandwidth on every transit link.
pub fn update_report_interval_with_bounds(&mut self, srtt_us: i64, min_ms: u64, max_ms: u64) {
if srtt_us <= 0 {
return;
}
let interval_ms = ((srtt_us as u64) / 1000).clamp(min_ms, max_ms);
self.report_interval_ms = interval_ms;
}
// --- Accessors ---
pub fn cumulative_packets_recv(&self) -> u64 {
self.cumulative_packets_recv
}
pub fn cumulative_bytes_recv(&self) -> u64 {
self.cumulative_bytes_recv
}
pub fn highest_counter(&self) -> u64 {
self.highest_counter
}
pub fn jitter_us(&self) -> u32 {
self.jitter.jitter_us()
}
pub fn report_interval_ms(&self) -> u64 {
self.report_interval_ms
}
/// Local monotonic time (injected `u64` ms) of the most recent received
/// frame, or `None` if none has been received.
pub fn last_recv_ms(&self) -> Option<u64> {
self.last_recv_ms
}
pub fn ecn_ce_count(&self) -> u32 {
self.ecn_ce_count
}
}
impl Default for ReceiverState {
fn default() -> Self {
Self::new(DEFAULT_OWD_WINDOW_SIZE)
}
}
src/proto/mmp/sender.rs
+213
View File
@@ -0,0 +1,213 @@
//! Per-peer sender-side MMP state (sans-IO).
//!
//! Records cumulative and interval TX counters and produces `SenderReport`
//! snapshots on demand. All time inputs are injected `u64` milliseconds.
use super::wire::SenderReport;
use super::{
COLD_START_SAMPLES, DEFAULT_COLD_START_INTERVAL_MS, MAX_REPORT_INTERVAL_MS,
MIN_REPORT_INTERVAL_MS,
};
/// Per-peer sender-side MMP state.
///
/// Records cumulative and interval counters for every frame transmitted
/// to this peer. Produces `SenderReport` snapshots on demand.
pub struct SenderState {
// --- Cumulative (lifetime) ---
cumulative_packets_sent: u64,
cumulative_bytes_sent: u64,
// --- Current interval ---
interval_start_counter: u64,
interval_start_timestamp: 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,
// --- Report timing (injected `u64` ms) ---
last_report_ms: Option<u64>,
report_interval_ms: u64,
// --- Send failure backoff ---
/// Consecutive send failure count for backoff calculation.
consecutive_send_failures: u32,
// --- Cold-start tracking ---
/// Number of SRTT-based interval updates received.
srtt_sample_count: u32,
}
impl SenderState {
pub fn new() -> Self {
Self::new_with_cold_start(DEFAULT_COLD_START_INTERVAL_MS)
}
/// Create with a custom cold-start interval (ms).
///
/// Used by session-layer MMP which needs a longer initial interval
/// since reports consume bandwidth on every transit link.
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_bytes_sent: 0,
last_counter: 0,
last_timestamp: 0,
interval_has_data: false,
last_report_ms: None,
report_interval_ms: cold_start_ms,
consecutive_send_failures: 0,
srtt_sample_count: 0,
}
}
/// Record a frame sent to this peer.
///
/// 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;
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.
///
/// Returns `None` if no frames have been sent since the last report.
/// `now_ms` is the injected monotonic time in milliseconds.
pub fn build_report(&mut self, now_ms: u64) -> Option<SenderReport> {
if !self.interval_has_data {
return None;
}
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_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_bytes_sent = 0;
self.last_report_ms = Some(now_ms);
Some(report)
}
/// Check if it's time to send a report.
///
/// When consecutive send failures have occurred, the effective interval
/// is multiplied by an exponential backoff factor (2^failures, capped at 32×).
pub fn should_send_report(&self, now_ms: u64) -> bool {
if !self.interval_has_data {
return false;
}
match self.last_report_ms {
None => true, // Never sent a report — send immediately
Some(last) => {
let effective_ms = (self.report_interval_ms as f64
* self.send_failure_backoff_multiplier())
as u64;
now_ms.saturating_sub(last) >= effective_ms
}
}
}
/// Record a send failure. Returns the new consecutive failure count.
pub fn record_send_failure(&mut self) -> u32 {
self.consecutive_send_failures += 1;
self.consecutive_send_failures
}
/// Record a successful send. Returns the previous failure count (for summary logging).
pub fn record_send_success(&mut self) -> u32 {
let prev = self.consecutive_send_failures;
self.consecutive_send_failures = 0;
prev
}
/// Get the backoff multiplier based on consecutive failures.
///
/// Returns 1.0 for no failures, 2.0 for 1 failure, 4.0 for 2, ...
/// capped at 32.0 (5 failures). Computed as an exact power-of-two bit shift
/// (`2^k == 1 << k`), keeping the module `core`-only (no `libm` for `powi`).
pub fn send_failure_backoff_multiplier(&self) -> f64 {
if self.consecutive_send_failures == 0 {
1.0
} else {
(1u64 << self.consecutive_send_failures.min(5)) as f64
}
}
/// Update the report interval based on SRTT (link-layer defaults).
///
/// Sender reports at 2× SRTT clamped to [floor, MAX]. During cold-start
/// (first `COLD_START_SAMPLES` updates), the floor is the cold-start
/// interval (200ms) for fast SRTT convergence. After that, it rises to
/// `MIN_REPORT_INTERVAL_MS` (1000ms) for steady-state efficiency.
pub fn update_report_interval_from_srtt(&mut self, srtt_us: i64) {
self.srtt_sample_count = self.srtt_sample_count.saturating_add(1);
let floor = if self.srtt_sample_count <= COLD_START_SAMPLES {
DEFAULT_COLD_START_INTERVAL_MS
} else {
MIN_REPORT_INTERVAL_MS
};
self.update_report_interval_with_bounds(srtt_us, floor, MAX_REPORT_INTERVAL_MS);
}
/// Update the report interval based on SRTT with custom bounds.
///
/// Used by session-layer MMP which needs higher clamp values since
/// each report consumes bandwidth on every transit link.
pub fn update_report_interval_with_bounds(&mut self, srtt_us: i64, min_ms: u64, max_ms: u64) {
if srtt_us <= 0 {
return;
}
let interval_us = (srtt_us * 2) as u64;
let interval_ms = (interval_us / 1000).clamp(min_ms, max_ms);
self.report_interval_ms = interval_ms;
}
// --- Accessors ---
pub fn cumulative_packets_sent(&self) -> u64 {
self.cumulative_packets_sent
}
pub fn cumulative_bytes_sent(&self) -> u64 {
self.cumulative_bytes_sent
}
pub fn report_interval_ms(&self) -> u64 {
self.report_interval_ms
}
pub fn consecutive_send_failures(&self) -> u32 {
self.consecutive_send_failures
}
}
impl Default for SenderState {
fn default() -> Self {
Self::new()
}
}
src/proto/mmp/state.rs
+13 -1156
View File
File diff suppressed because it is too large Load Diff
src/proto/mmp/tests/state.rs
+2 -1
View File
@@ -1,6 +1,7 @@
//! Tests for the owned MMP protocol state machines.
use crate::proto::mmp::state::{ReceiverState, SenderState};
use crate::proto::mmp::receiver::ReceiverState;
use crate::proto::mmp::sender::SenderState;
use crate::proto::mmp::{
COLD_START_SAMPLES, DEFAULT_LOG_INTERVAL_SECS, DEFAULT_OWD_WINDOW_SIZE, MAX_REPORT_INTERVAL_MS,
MIN_REPORT_INTERVAL_MS, MmpMetrics, MmpMode, ReceiverReport,
src/proto/stp/limits.rs
+1 -1
View File
@@ -11,7 +11,7 @@
//! `flap_dampening_duration`.
//!
//! The monotonic clock is injected: every timing method takes a `now_ms: u64`
//! (monotonic milliseconds, read by the shell from `crate::mmp::mono_ms`), and
//! (monotonic milliseconds, read by the shell from `crate::time::mono_ms`), and
//! all timers/durations are stored as plain `u64` milliseconds. Configuration
//! setters accept seconds (matching the node config) and store the value scaled
//! to milliseconds so the comparisons stay in one unit.
src/time.rs
+16
View File
@@ -0,0 +1,16 @@
//! Shell-side time seam: the process-monotonic millisecond clock the sans-IO
//! cores take as an injected `u64`.
use std::sync::LazyLock;
use std::time::Instant;
/// Monotonic millisecond clock for the MMP time seam.
///
/// The owned `proto::mmp` state takes injected `u64` milliseconds rather than
/// reading a clock. The shell reads this process-monotonic clock at each edge
/// and passes the value in; only deltas between two `mono_ms()` reads are ever
/// compared, so the (process-relative) epoch is immaterial.
pub(crate) fn mono_ms() -> u64 {
static START: LazyLock<Instant> = LazyLock::new(Instant::now);
START.elapsed().as_millis() as u64
}