diff --git a/src/config/mod.rs b/src/config/mod.rs index 0097a15..211254f 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -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::{ diff --git a/src/config/node.rs b/src/config/node.rs index ae50541..613ff7d 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -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(); diff --git a/src/lib.rs b/src/lib.rs index 9c1ea15..af83669 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -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; diff --git a/src/mmp/mod.rs b/src/mmp/mod.rs deleted file mode 100644 index 5a729bb..0000000 --- a/src/mmp/mod.rs +++ /dev/null @@ -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 = 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); - } -} diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 3df8a43..7401f9b 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -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(); diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index 9362c6c..39ea9c7 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -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; diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 952f5ef..3abc8e3 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -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!( diff --git a/src/node/session.rs b/src/node/session.rs index 694012e..18bcde2 100644 --- a/src/node/session.rs +++ b/src/node/session.rs @@ -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); } diff --git a/src/node/tests/mmp_chartests.rs b/src/node/tests/mmp_chartests.rs index f3e39d6..efd8db8 100644 --- a/src/node/tests/mmp_chartests.rs +++ b/src/node/tests/mmp_chartests.rs @@ -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" diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index 18dc267..b45297c 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -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); } diff --git a/src/node/tree.rs b/src/node/tree.rs index d595584..d7e7fc3 100644 --- a/src/node/tree.rs +++ b/src/node/tree.rs @@ -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 diff --git a/src/peer/active.rs b/src/peer/active.rs index 9c9aea4..c88e95a 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -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); } diff --git a/src/proto/mmp/core.rs b/src/proto/mmp/core.rs index d38398a..4d832bb 100644 --- a/src/proto/mmp/core.rs +++ b/src/proto/mmp/core.rs @@ -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. diff --git a/src/proto/mmp/limits.rs b/src/proto/mmp/limits.rs new file mode 100644 index 0000000..0ada036 --- /dev/null +++ b/src/proto/mmp/limits.rs @@ -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::` 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; diff --git a/src/proto/mmp/metrics.rs b/src/proto/mmp/metrics.rs new file mode 100644 index 0000000..6663f48 --- /dev/null +++ b/src/proto/mmp/metrics.rs @@ -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, + /// 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 { + 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 { + 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 { + 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() + } +} diff --git a/src/proto/mmp/mod.rs b/src/proto/mmp/mod.rs index 37db794..400dbe8 100644 --- a/src/proto/mmp/mod.rs +++ b/src/proto/mmp/mod.rs @@ -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::` 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, +}; diff --git a/src/proto/mmp/path_mtu.rs b/src/proto/mmp/path_mtu.rs new file mode 100644 index 0000000..f9b676d --- /dev/null +++ b/src/proto/mmp/path_mtu.rs @@ -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, + /// 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, + /// 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 { + 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() + } +} diff --git a/src/proto/mmp/receiver.rs b/src/proto/mmp/receiver.rs new file mode 100644 index 0000000..15b9f44 --- /dev/null +++ b/src/proto/mmp/receiver.rs @@ -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, + /// 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, + + // --- 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, + + // --- Report timing (injected `u64` ms) --- + last_report_ms: Option, + 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 { + 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 { + 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) + } +} diff --git a/src/proto/mmp/sender.rs b/src/proto/mmp/sender.rs new file mode 100644 index 0000000..6c98158 --- /dev/null +++ b/src/proto/mmp/sender.rs @@ -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, + 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 { + 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() + } +} diff --git a/src/proto/mmp/state.rs b/src/proto/mmp/state.rs index c4184a3..5730a17 100644 --- a/src/proto/mmp/state.rs +++ b/src/proto/mmp/state.rs @@ -1,12 +1,11 @@ -//! Owned MMP protocol state (sans-IO). +//! Owned MMP protocol state anchor and per-peer/per-session aggregates (sans-IO). //! -//! The per-peer/per-session sender, receiver, derived-metrics, spin-bit and -//! path-MTU state machines migrated out of the async node shell. All time -//! inputs are injected `u64` milliseconds (the shell reads its monotonic clock -//! at the edge and passes the value in); nothing here reads a clock, logs, or -//! bumps a counter. `no_std`+`alloc`-clean: `core` arithmetic only, the ring -//! buffer comes from `super::algorithms` (which uses `alloc`), and observability -//! is returned to the shell as an [`RrLog`] outcome rather than emitted here. +//! [`Mmp`] is the stateless reporting anchor owned by `Node`; the per-entity +//! aggregates [`MmpPeerState`]/[`MmpSessionState`] wrap the owned role state +//! machines that live in sibling modules (`sender`, `receiver`, `metrics`, +//! `path_mtu`). All time inputs are injected `u64` milliseconds (the shell reads +//! its monotonic clock at the edge and passes the value in); nothing here reads a +//! clock, logs, or bumps a counter. `no_std`+`alloc`-clean. //! //! [`Mmp`] itself remains the stateless reporting anchor owned by `Node` (the //! live per-entity state lives on the peers'/sessions' shell structs); it exists @@ -15,22 +14,12 @@ use core::fmt::{self, Debug}; -use super::algorithms::{ - DualEwma, JitterEstimator, OwdTrendDetector, SpinBitState, SrttEstimator, compute_etx, -}; -use super::wire::{ReceiverReport, SenderReport}; -use super::{ - COLD_START_SAMPLES, DEFAULT_COLD_START_INTERVAL_MS, DEFAULT_OWD_WINDOW_SIZE, - MAX_REPORT_INTERVAL_MS, MIN_REPORT_INTERVAL_MS, MmpMode, SESSION_COLD_START_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; +use super::algorithms::SpinBitState; +use super::metrics::MmpMetrics; +use super::path_mtu::PathMtuState; +use super::receiver::ReceiverState; +use super::sender::SenderState; +use super::{MmpMode, SESSION_COLD_START_INTERVAL_MS}; /// MMP reporting subsystem anchor owned by [`Node`](crate::node::Node). /// @@ -51,964 +40,6 @@ impl Mmp { } } -// ============================================================================ -// 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, -} - -// ============================================================================ -// SenderState -// ============================================================================ - -/// 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, - 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 { - 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() - } -} - -// ============================================================================ -// 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, - /// 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, - - // --- 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, - - // --- Report timing (injected `u64` ms) --- - last_report_ms: Option, - 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 { - 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 { - 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) - } -} - -// ============================================================================ -// 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, - /// 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 { - 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 { - 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 { - 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() - } -} - // ============================================================================ // Per-Peer MMP State // ============================================================================ @@ -1163,177 +194,3 @@ impl Debug for MmpSessionState { .finish_non_exhaustive() } } - -// ============================================================================ -// Path MTU State (session-layer only) -// ============================================================================ - -/// 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, - /// 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, - /// 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 { - 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() - } -} diff --git a/src/proto/mmp/tests/state.rs b/src/proto/mmp/tests/state.rs index 790cf77..378234a 100644 --- a/src/proto/mmp/tests/state.rs +++ b/src/proto/mmp/tests/state.rs @@ -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, diff --git a/src/proto/stp/limits.rs b/src/proto/stp/limits.rs index 9fc36ca..033a0bd 100644 --- a/src/proto/stp/limits.rs +++ b/src/proto/stp/limits.rs @@ -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. diff --git a/src/time.rs b/src/time.rs new file mode 100644 index 0000000..d684f47 --- /dev/null +++ b/src/time.rs @@ -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 = LazyLock::new(Instant::now); + START.elapsed().as_millis() as u64 +}