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

Merge refactor-sans-io: MMP sans-IO reporting on the next line

Browse Source
This commit is contained in:
Johnathan Corgan
2026-07-07 11:30:40 +00:00
parent 3b46bbe17d 50a595a0ed
commit 584edfac20
28 changed files with 4551 additions and 3043 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/decrypt_worker.rs
+3 -4
View File
@@ -112,8 +112,8 @@ pub(crate) struct DecryptJob {
pub fmp_counter: u64,
/// Flag byte from the FMP outer header. Carried through the
/// fallback so the rx_loop bounce arm can extract `CE` and `SP`
/// for ECN propagation, MMP stats, and spin-bit RTT
/// observation — these used to be dropped on the worker path
/// for ECN propagation and MMP stats — these used to be dropped
/// on the worker path
/// because the bounce hardcoded `fmp_flags: 0`.
pub fmp_flags: u8,
/// 16-byte FMP outer header used as AAD during AEAD open.
@@ -544,8 +544,7 @@ mod tests {
/// `DecryptFallback.fmp_flags`. Pre-fix the worker hardcoded
/// `fmp_flags: 0`, dropping CE / SP on every packet handled by
/// the production worker path (i.e. every bulk-data packet).
/// Loss of CE wrecks ECN propagation; loss of SP wrecks
/// spin-bit RTT observation.
/// Loss of CE wrecks ECN propagation.
///
/// Drives the worker's `handle_job` directly: build an FMP wire
/// packet sealed with a known cipher, ship a `DecryptJob` with
src/node/handlers/encrypted.rs
+6 -7
View File
@@ -4,7 +4,6 @@ use crate::node::Node;
use crate::node::wire::{EncryptedHeader, FLAG_CE, FLAG_KEY_EPOCH, strip_inner_header};
use crate::noise::NoiseError;
use crate::transport::ReceivedPacket;
use std::time::Instant;
use tracing::{debug, trace, warn};
/// Force-remove a peer after this many consecutive decryption failures.
@@ -257,7 +256,7 @@ impl Node {
};
// MMP per-frame processing and statistics
let now = Instant::now();
let now_ms = crate::mmp::mono_ms();
let ce_flag = header.flags & FLAG_CE != 0;
if let Some(peer) = self.peers.get_mut(&node_addr) {
@@ -274,7 +273,7 @@ impl Node {
timestamp,
packet.data.len(),
ce_flag,
now,
now_ms,
);
}
peer.set_current_addr(packet.transport_id, packet.remote_addr.clone());
@@ -331,8 +330,8 @@ impl Node {
/// Canonical post-FMP-decrypt side-effect site. Used by both the
/// inline rx_loop decrypt path and the decrypt-worker bounce path
/// so the per-peer bookkeeping (stats, MMP, spin-bit RTT, ECN
/// propagation, address-rotation handling, link-message dispatch)
/// so the per-peer bookkeeping (stats, MMP, ECN propagation,
/// address-rotation handling, link-message dispatch)
/// happens in exactly one place.
#[allow(clippy::too_many_arguments)]
pub(in crate::node) async fn process_authentic_fmp_plaintext(
@@ -357,7 +356,7 @@ impl Node {
} else {
return;
};
let now = Instant::now();
let now_ms = crate::mmp::mono_ms();
let mut address_changed = false;
if let Some(peer) = self.peers.get_mut(node_addr) {
peer.reset_decrypt_failures();
@@ -379,7 +378,7 @@ impl Node {
peer.touch(packet_timestamp_ms);
if let Some(mmp) = peer.mmp_mut() {
mmp.receiver
.record_recv(fmp_counter, inner_ts, packet_len, ce_flag, now);
.record_recv(fmp_counter, inner_ts, packet_len, ce_flag, now_ms);
}
}
// Address rotation invalidates the per-peer connect()-ed UDP
src/node/handlers/mmp.rs
+348 -248
View File
@@ -5,18 +5,57 @@
//! and teardown metric logs.
use crate::NodeAddr;
use crate::mmp::MmpMode;
use crate::mmp::MmpSessionState;
use crate::mmp::report::{ReceiverReport, SenderReport};
use crate::node::Node;
use crate::node::reject::{MmpReject, RejectReason, TreeReject};
use crate::protocol::{
LinkMessageType, PathMtuNotification, SessionMessageType, SessionReceiverReport,
SessionSenderReport,
use crate::proto::mmp::{
BackoffUpdate, LinkReportKind, LinkReportSnapshot, MmpAction, MmpSessionState,
PathMtuNotification, PeerLivenessSnapshot, ReceiverReport, RrLog, SendResult, SenderReport,
SessionReceiverReport, SessionReportKind, SessionReportSnapshot, SessionSenderReport,
};
use crate::protocol::{LinkMessageType, SessionMessageType};
use std::time::{Duration, Instant};
use tracing::{debug, info, trace, warn};
/// Emit the operator `trace!` point for a processed ReceiverReport outcome.
///
/// These log points used to live inside `MmpMetrics::process_receiver_report`;
/// the sans-IO migration returns the outcome as an [`RrLog`] and re-emits it
/// here, shell-side, preserving the original field set, content, and (relative
/// to the surrounding handler logs) ordering. The original traces carried no
/// peer identifier, so none is added here.
pub(super) fn log_rr_outcome(rr: &ReceiverReport, our_timestamp_ms: u32, log: RrLog) {
match log {
RrLog::Stale {
prev_highest,
prev_packets,
prev_bytes,
} => trace!(
highest_counter = rr.highest_counter,
prev_highest_counter = prev_highest,
cumulative_packets_recv = rr.cumulative_packets_recv,
prev_cumulative_packets_recv = prev_packets,
cumulative_bytes_recv = rr.cumulative_bytes_recv,
prev_cumulative_bytes_recv = prev_bytes,
"Ignoring stale MMP ReceiverReport"
),
RrLog::RttSample { rtt_ms, srtt_ms } => trace!(
our_ts = our_timestamp_ms,
echo = rr.timestamp_echo,
dwell = u32::from(rr.dwell_time),
rtt_ms = rtt_ms,
srtt_ms = srtt_ms,
"RTT sample from timestamp echo"
),
RrLog::InvalidRtt => trace!(
our_ts = our_timestamp_ms,
echo = rr.timestamp_echo,
dwell = u32::from(rr.dwell_time),
"Ignoring invalid MMP RTT sample"
),
RrLog::None => {}
}
}
/// Format bytes/sec as human-readable throughput.
fn format_throughput(bps: f64) -> String {
if bps == 0.0 {
@@ -114,10 +153,12 @@ impl Node {
// Process the report: computes RTT from timestamp echo, updates
// loss rate, goodput rate, jitter trend, and ETX.
let now = Instant::now();
let first_rtt = mmp
.metrics
.process_receiver_report(&rr, our_timestamp_ms, now);
let now_ms = crate::mmp::mono_ms();
let (first_rtt, rr_log) =
mmp.metrics
.process_receiver_report(&rr, our_timestamp_ms, now_ms);
// Re-emit the operator trace the core used to log mid-decision.
log_rr_outcome(&rr, our_timestamp_ms, rr_log);
// Feed SRTT back to sender/receiver report interval tuning
if let Some(srtt_ms) = mmp.metrics.srtt_ms() {
@@ -225,70 +266,91 @@ 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 = Instant::now();
let now_ms = crate::mmp::mono_ms();
// Collect peers that need reports (can't borrow self mutably while iterating)
let mut sender_reports: Vec<(NodeAddr, Vec<u8>)> = Vec::new();
let mut receiver_reports: Vec<(NodeAddr, Vec<u8>)> = Vec::new();
// Build one report-gating snapshot per peer, resolving every timing read
// shell-side into a `bool`. `send_sr`/`send_rr` come from the peer's
// negotiated profile (whether it provides/wants each report); the core
// ANDs them into the mode/timing gate. The snapshots own only
// `NodeAddr`/`MmpMode`/`bool`, so the peer-iteration borrow is released
// before the pure decision runs and the driving loop mutates the
// reporting state.
let snapshots: Vec<LinkReportSnapshot> = self
.peers
.iter()
.filter_map(|(node_addr, peer)| {
let mmp = peer.mmp()?;
Some(LinkReportSnapshot {
peer: *node_addr,
mode: mmp.mode(),
send_sr: peer.send_sr(),
send_rr: peer.send_rr(),
sr_due: mmp.sender.should_send_report(now_ms),
rr_due: mmp.receiver.should_send_report(now_ms),
log_due: mmp.should_log(now_ms),
})
})
.collect();
for (node_addr, peer) in self.peers.iter_mut() {
// Compute display name before taking mutable MMP borrow
let peer_name = self
.peer_aliases
.get(node_addr)
.cloned()
.unwrap_or_else(|| peer.identity().short_npub());
let actions = self.mmp.plan_link_reports(&snapshots);
let send_sr = peer.send_sr();
let send_rr = peer.send_rr();
let Some(mmp) = peer.mmp_mut() else {
continue;
};
let mode = mmp.mode();
// Sender reports: gated by mode, profile wants/provides, and timing
if mode == MmpMode::Full
&& send_sr
&& mmp.sender.should_send_report(now)
&& let Some(sr) = mmp.sender.build_report(now)
{
sender_reports.push((*node_addr, sr.encode()));
}
// Receiver reports: gated by mode, profile wants/provides, and timing
if mode != MmpMode::Minimal
&& send_rr
&& mmp.receiver.should_send_report(now)
&& let Some(rr) = mmp.receiver.build_report(now)
{
receiver_reports.push((*node_addr, rr.encode()));
}
// Periodic operator logging
if mmp.should_log(now) {
Self::log_mmp_metrics(&peer_name, mmp);
mmp.mark_logged(now);
}
}
// Send collected reports
for (node_addr, encoded) in sender_reports {
if let Err(e) = self.send_encrypted_link_message(&node_addr, &encoded).await {
debug!(peer = %self.peer_display_name(&node_addr), error = %e, "Failed to send SenderReport");
}
}
for (node_addr, encoded) in receiver_reports {
if let Err(e) = self.send_encrypted_link_message(&node_addr, &encoded).await {
debug!(peer = %self.peer_display_name(&node_addr), error = %e, "Failed to send ReceiverReport");
// Drive the planned actions in their phase-grouped order (all logs, then
// all SenderReports, then all ReceiverReports). Logs run first because the
// operator log reads cumulative_packets_sent, which each report send
// advances (send_encrypted_link_message -> sender.record_sent); the
// pre-refactor handler logged during its collect pass, before any send.
// `build_report` (which advances the interval state) is called only on a
// SendLinkReport action, exactly as the pre-refactor gate did.
for action in actions {
match action {
MmpAction::SendLinkReport { peer, kind } => {
let encoded = self
.peers
.get_mut(&peer)
.and_then(|p| p.mmp_mut())
.and_then(|mmp| match kind {
LinkReportKind::Sender => {
mmp.sender.build_report(now_ms).map(|sr| sr.encode())
}
LinkReportKind::Receiver => {
mmp.receiver.build_report(now_ms).map(|rr| rr.encode())
}
});
if let Some(encoded) = encoded
&& let Err(e) = self.send_encrypted_link_message(&peer, &encoded).await
{
let label = match kind {
LinkReportKind::Sender => "Failed to send SenderReport",
LinkReportKind::Receiver => "Failed to send ReceiverReport",
};
debug!(peer = %self.peer_display_name(&peer), error = %e, "{}", label);
}
}
MmpAction::LogLink { peer } => {
// Resolve the display name exactly as the pre-refactor loop
// did (alias, else short_npub) — not `peer_display_name`,
// which also consults the host map.
let peer_name = self.peer_aliases.get(&peer).cloned().unwrap_or_else(|| {
self.peers
.get(&peer)
.map(|p| p.identity().short_npub())
.unwrap_or_default()
});
if let Some(mmp) = self.peers.get_mut(&peer).and_then(|p| p.mmp_mut()) {
Self::log_mmp_metrics(&peer_name, mmp);
mmp.mark_logged(now_ms);
}
}
MmpAction::ReapPeer { .. }
| MmpAction::Heartbeat { .. }
| MmpAction::SendSessionReport { .. }
| MmpAction::LogSession { .. } => {}
}
}
}
/// Emit periodic MMP metrics for a peer.
fn log_mmp_metrics(peer_name: &str, mmp: &crate::mmp::MmpPeerState) {
fn log_mmp_metrics(peer_name: &str, mmp: &crate::proto::mmp::MmpPeerState) {
let m = &mmp.metrics;
let rtt_str = if m.rtt_trend.initialized() {
@@ -316,7 +378,10 @@ impl Node {
}
/// Emit a teardown log summarizing lifetime MMP metrics for a removed peer.
pub(in crate::node) fn log_mmp_teardown(peer_name: &str, mmp: &crate::mmp::MmpPeerState) {
pub(in crate::node) fn log_mmp_teardown(
peer_name: &str,
mmp: &crate::proto::mmp::MmpPeerState,
) {
let m = &mmp.metrics;
let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;
@@ -347,136 +412,154 @@ 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 = Instant::now();
let now_ms = crate::mmp::mono_ms();
// Collect reports to send: (dest_addr, msg_type, encoded_body)
let mut reports: Vec<(NodeAddr, u8, Vec<u8>)> = Vec::new();
// Build one report-gating snapshot per session, resolving every timing
// read shell-side into a `bool`. The snapshots own only
// `NodeAddr`/`MmpMode`/`bool`, so the session-iteration borrow is released
// before the pure decision runs and the driving loop mutates the
// reporting state / performs the sends.
let snapshots: Vec<SessionReportSnapshot> = self
.sessions
.iter()
.filter_map(|(dest_addr, entry)| {
let mmp = entry.mmp()?;
Some(SessionReportSnapshot {
dest: *dest_addr,
mode: mmp.mode(),
sr_due: mmp.sender.should_send_report(now_ms),
rr_due: mmp.receiver.should_send_report(now_ms),
mtu_due: mmp.path_mtu.should_send_notification(now_ms),
log_due: mmp.should_log(now_ms),
})
})
.collect();
for (dest_addr, entry) in self.sessions.iter_mut() {
// Compute display name before taking mutable MMP borrow
let session_name = self
.peer_aliases
.get(dest_addr)
.cloned()
.unwrap_or_else(|| {
let (xonly, _) = entry.remote_pubkey().x_only_public_key();
crate::PeerIdentity::from_pubkey(xonly).short_npub()
});
let actions = self.mmp.plan_session_reports(&snapshots);
let Some(mmp) = entry.mmp_mut() else {
continue;
};
let mode = mmp.mode();
// Sender reports: Full mode only
if mode == MmpMode::Full
&& mmp.sender.should_send_report(now)
&& let Some(sr) = mmp.sender.build_report(now)
{
let session_sr: SessionSenderReport = SessionSenderReport::from(&sr);
reports.push((
*dest_addr,
SessionMessageType::SenderReport.to_byte(),
session_sr.encode(),
));
}
// Receiver reports: Full and Lightweight modes
if mode != MmpMode::Minimal
&& mmp.receiver.should_send_report(now)
&& let Some(rr) = mmp.receiver.build_report(now)
{
let session_rr: SessionReceiverReport = SessionReceiverReport::from(&rr);
reports.push((
*dest_addr,
SessionMessageType::ReceiverReport.to_byte(),
session_rr.encode(),
));
}
// PathMtu notifications (all modes)
if mmp.path_mtu.should_send_notification(now)
&& let Some(mtu_value) = mmp.path_mtu.build_notification(now)
{
let notif = PathMtuNotification::new(mtu_value);
reports.push((
*dest_addr,
SessionMessageType::PathMtuNotification.to_byte(),
notif.encode(),
));
}
// Periodic operator logging
if mmp.should_log(now) {
Self::log_session_mmp_metrics(&session_name, mmp);
mmp.mark_logged(now);
}
}
// Send collected reports via session-layer encryption.
// Track per-destination success/failure for backoff and log suppression.
let mut send_results: Vec<(NodeAddr, bool)> = Vec::new();
for (dest_addr, msg_type, body) in reports {
match self.send_session_msg(&dest_addr, msg_type, &body).await {
Ok(()) => {
send_results.push((dest_addr, true));
// Drive the planned actions in phase-grouped order (all logs, then the
// sends in per-session SR/RR/MTU order). Logs run first because the
// session operator log reads cumulative_packets_sent, which each send
// advances (send_session_msg -> sender.record_sent); the pre-refactor
// handler logged during its collect pass, before any send. Each build
// (`build_report`/`build_notification`, which advance interval/
// notification state) runs only on its SendSessionReport action, exactly
// as the pre-refactor collect pass did. Per-destination success/failure
// is collected for the backoff dedup + failure-log suppression.
let mut send_results: Vec<SendResult> = Vec::new();
for action in actions {
match action {
MmpAction::LogSession { dest } => {
// Resolve the display name exactly as the pre-refactor loop
// did (alias, else short_npub from the session's remote key).
let session_name = self.peer_aliases.get(&dest).cloned().unwrap_or_else(|| {
self.sessions
.get(&dest)
.map(|entry| {
let (xonly, _) = entry.remote_pubkey().x_only_public_key();
crate::PeerIdentity::from_pubkey(xonly).short_npub()
})
.unwrap_or_default()
});
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
Self::log_session_mmp_metrics(&session_name, mmp);
mmp.mark_logged(now_ms);
}
}
Err(e) => {
// Peek at current failure count for log suppression
let failures = self
MmpAction::SendSessionReport { dest, kind } => {
let built = self
.sessions
.get(&dest_addr)
.and_then(|entry| entry.mmp())
.map(|mmp| mmp.sender.consecutive_send_failures())
.unwrap_or(0);
.get_mut(&dest)
.and_then(|entry| entry.mmp_mut())
.and_then(|mmp| match kind {
SessionReportKind::Sender => {
mmp.sender.build_report(now_ms).map(|sr| {
(
SessionMessageType::SenderReport.to_byte(),
SessionSenderReport::from(&sr).encode(),
)
})
}
SessionReportKind::Receiver => {
mmp.receiver.build_report(now_ms).map(|rr| {
(
SessionMessageType::ReceiverReport.to_byte(),
SessionReceiverReport::from(&rr).encode(),
)
})
}
SessionReportKind::PathMtu => {
mmp.path_mtu.build_notification(now_ms).map(|mtu_value| {
(
SessionMessageType::PathMtuNotification.to_byte(),
PathMtuNotification::new(mtu_value).encode(),
)
})
}
});
if failures < 3 {
debug!(
dest = %self.peer_display_name(&dest_addr),
msg_type,
error = %e,
"Failed to send session MMP report"
);
} else if failures == 3 {
debug!(
dest = %self.peer_display_name(&dest_addr),
"Suppressing further session MMP send failure logs"
);
let Some((msg_type, body)) = built else {
continue;
};
match self.send_session_msg(&dest, msg_type, &body).await {
Ok(()) => send_results.push(SendResult { dest, ok: true }),
Err(e) => {
// Peek at current failure count for log suppression
// (unchanged by the backoff apply, which runs later).
let failures = self
.sessions
.get(&dest)
.and_then(|entry| entry.mmp())
.map(|mmp| mmp.sender.consecutive_send_failures())
.unwrap_or(0);
if failures < 3 {
debug!(
dest = %self.peer_display_name(&dest),
msg_type,
error = %e,
"Failed to send session MMP report"
);
} else if failures == 3 {
debug!(
dest = %self.peer_display_name(&dest),
"Suppressing further session MMP send failure logs"
);
}
// failures > 3: silently suppressed
send_results.push(SendResult { dest, ok: false });
}
}
// failures > 3: silently suppressed
send_results.push((dest_addr, false));
}
MmpAction::ReapPeer { .. }
| MmpAction::Heartbeat { .. }
| MmpAction::SendLinkReport { .. }
| MmpAction::LogLink { .. } => {}
}
}
// Update backoff state from send results.
// Deduplicate: a destination counts as success if ANY report succeeded,
// failure only if ALL reports for that destination failed.
let mut dest_success: std::collections::HashMap<NodeAddr, bool> =
std::collections::HashMap::new();
for (dest, ok) in &send_results {
let entry = dest_success.entry(*dest).or_insert(false);
if *ok {
*entry = true;
}
}
for (dest_addr, success) in dest_success {
if let Some(entry) = self.sessions.get_mut(&dest_addr)
&& let Some(mmp) = entry.mmp_mut()
{
if success {
let prev = mmp.sender.record_send_success();
if prev > 3 {
debug!(
dest = %self.peer_display_name(&dest_addr),
consecutive_failures = prev,
"Resumed session MMP reporting"
);
// Deduplicate send results per destination (any-ok -> success, all-fail
// -> failure) and apply the backoff state transition for each dest.
for update in self.mmp.plan_backoff(&send_results) {
match update {
BackoffUpdate::Success { dest } => {
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
let prev = mmp.sender.record_send_success();
if prev > 3 {
debug!(
dest = %self.peer_display_name(&dest),
consecutive_failures = prev,
"Resumed session MMP reporting"
);
}
}
}
BackoffUpdate::Failure { dest } => {
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
mmp.sender.record_send_failure();
}
} else {
mmp.sender.record_send_failure();
}
}
}
@@ -545,84 +628,101 @@ impl Node {
/// hasn't sent us a frame within the link dead timeout.
pub(in crate::node) async fn check_link_heartbeats(&mut self) {
let now = Instant::now();
// 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 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;
let max_resends = self.config().node.rate_limit.handshake_max_resends;
let heartbeat_msg = [LinkMessageType::Heartbeat.to_byte()];
// Collect heartbeats to send and dead peers to remove
let mut heartbeats: Vec<NodeAddr> = Vec::new();
let mut dead_peers: Vec<NodeAddr> = Vec::new();
// Build one liveness snapshot per peer, resolving every clock read and
// the rekey-suppression predicate shell-side. The snapshots own only
// `NodeAddr`/`bool`, so the peer-iteration borrow is released before the
// pure decision runs and the driving loop mutates the registry.
let snapshots: Vec<PeerLivenessSnapshot> = self
.peers
.iter()
.map(|(node_addr, peer)| {
// Check liveness via the MMP receiver's last-received monotonic
// ms. Fall back to session_start (an `ActivePeer` Instant) for
// peers that never sent data, keeping that branch in Instant
// space so no monotonic-ms epoch conversion is needed.
let time_dead = if let Some(mmp) = peer.mmp() {
match mmp.receiver.last_recv_ms() {
Some(last_ms) => now_ms.saturating_sub(last_ms) >= dead_timeout_ms,
None => now.duration_since(peer.session_start()) >= dead_timeout,
}
} else {
false
};
for (node_addr, peer) in self.peers.iter() {
// Check liveness via MMP receiver last_recv_time.
// Fall back to session_start for peers that never sent data.
let time_dead = if let Some(mmp) = peer.mmp() {
let reference_time = mmp
.receiver
.last_recv_time()
.unwrap_or(peer.session_start());
now.duration_since(reference_time) >= dead_timeout
} else {
false
};
// Suppress teardown while an FMP rekey is genuinely in flight
// with budget left: a rekey-handshake link is not silent —
// whether mid-msg1 or mid-msg3 retransmit. The resend caps
// guarantee this terminates (abandon on exhaustion or cutover on
// completion clears the rekey state), so a truly dead link is
// reaped on the next cycle.
let rekey_active = (peer.rekey_in_progress()
&& peer.rekey_msg1_resend_count() < max_resends
&& peer.rekey_msg1().is_some())
|| (peer.rekey_msg3_payload().is_some()
&& peer.rekey_msg3_resend_count() < max_resends);
// Suppress teardown while an FMP rekey is genuinely in flight with
// budget left: a rekey-handshake link is not silent. The msg1
// resend cap guarantees this terminates (abandon on exhaustion or
// cutover on completion clears `rekey_in_progress`), so a truly
// dead link is reaped on the next cycle.
let rekey_active = peer.rekey_in_progress()
&& peer.rekey_msg1_resend_count() < max_resends
&& peer.rekey_msg1().is_some()
|| (peer.rekey_msg3_payload().is_some()
&& peer.rekey_msg3_resend_count() < max_resends);
// Check if heartbeat is due.
let heartbeat_due = match peer.last_heartbeat_sent() {
None => true,
Some(last) => now.duration_since(last) >= heartbeat_interval,
};
let is_dead = time_dead && !rekey_active;
if is_dead {
dead_peers.push(*node_addr);
continue;
}
PeerLivenessSnapshot {
peer: *node_addr,
time_dead,
rekey_active,
heartbeat_due,
}
})
.collect();
// Check if heartbeat is due
let needs_heartbeat = match peer.last_heartbeat_sent() {
None => true,
Some(last) => now.duration_since(last) >= heartbeat_interval,
};
if needs_heartbeat {
heartbeats.push(*node_addr);
}
}
let actions = self.mmp.plan_heartbeats(&snapshots);
// Remove dead peers and schedule auto-reconnect
// Wall-clock basis for reconnect scheduling, sourced once (as before).
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
for addr in &dead_peers {
debug!(
peer = %self.peer_display_name(addr),
timeout_secs = self.config().node.link_dead_timeout_secs,
"Removing peer: link dead timeout"
);
self.remove_active_peer(addr);
self.schedule_reconnect(*addr, now_ms);
}
// Send heartbeats (skip peers we just removed)
for addr in heartbeats {
if dead_peers.contains(&addr) {
continue;
}
if let Some(peer) = self.peers.get_mut(&addr) {
peer.mark_heartbeat_sent(now);
}
if let Err(e) = self
.send_encrypted_link_message(&addr, &heartbeat_msg)
.await
{
trace!(peer = %self.peer_display_name(&addr), error = %e, "Failed to send heartbeat");
// Drive the planned actions: all reaps first (each removed +
// reconnect-scheduled), then all heartbeats (a just-reaped peer is never
// heartbeated — the core never emits both for the same peer).
for action in actions {
match action {
MmpAction::ReapPeer { peer } => {
debug!(
peer = %self.peer_display_name(&peer),
timeout_secs = self.config().node.link_dead_timeout_secs,
"Removing peer: link dead timeout"
);
self.remove_active_peer(&peer);
self.schedule_reconnect(peer, now_ms);
}
MmpAction::Heartbeat { peer } => {
if let Some(p) = self.peers.get_mut(&peer) {
p.mark_heartbeat_sent(now);
}
if let Err(e) = self
.send_encrypted_link_message(&peer, &heartbeat_msg)
.await
{
trace!(peer = %self.peer_display_name(&peer), error = %e, "Failed to send heartbeat");
}
}
MmpAction::SendLinkReport { .. }
| MmpAction::LogLink { .. }
| MmpAction::SendSessionReport { .. }
| MmpAction::LogSession { .. } => {}
}
}
}
src/node/handlers/session.rs
+21 -15
View File
@@ -6,8 +6,6 @@
//! encrypted data, and error signals (CoordsRequired, PathBroken).
use crate::NodeAddr;
use crate::mmp::report::ReceiverReport;
use crate::mmp::{MAX_SESSION_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS};
use crate::node::reject::{RejectReason, SessionReject};
use crate::node::session::{EndToEndState, EpochSlot, SessionEntry};
use crate::node::session_wire::{
@@ -21,14 +19,17 @@ use crate::node::wire::{ESTABLISHED_HEADER_SIZE, FLAG_KEY_EPOCH, build_establish
use crate::node::{Node, NodeError};
use crate::noise::{HANDSHAKE_MSG1_SIZE, HANDSHAKE_MSG2_SIZE, HANDSHAKE_MSG3_SIZE, HandshakeState};
use crate::proto::fmp::NegotiationPayload;
use crate::proto::mmp::{MAX_SESSION_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS};
use crate::proto::mmp::{
PathMtuNotification, ReceiverReport, SessionReceiverReport, SessionSenderReport,
};
use crate::proto::routing::{CoordsRequired, MtuExceeded, PathBroken};
#[cfg(unix)]
use crate::protocol::LinkMessageType;
#[cfg(unix)]
use crate::protocol::SESSION_DATAGRAM_HEADER_SIZE;
use crate::protocol::{
FspInnerFlags, PathMtuNotification, SessionAck, SessionDatagram, SessionMessageType,
SessionMsg3, SessionReceiverReport, SessionSenderReport, SessionSetup,
FspInnerFlags, SessionAck, SessionDatagram, SessionMessageType, SessionMsg3, SessionSetup,
};
use crate::protocol::{coords_wire_size, encode_coords};
#[cfg(unix)]
@@ -302,9 +303,9 @@ impl Node {
if let Some(entry) = self.sessions.get_mut(src_addr)
&& let Some(mmp) = entry.mmp_mut()
{
let now = std::time::Instant::now();
let now_ms = crate::mmp::mono_ms();
mmp.receiver
.record_recv(header.counter, timestamp, plaintext.len(), ce_flag, now);
.record_recv(header.counter, timestamp, plaintext.len(), ce_flag, now_ms);
let _inner_flags = FspInnerFlags::from_byte(inner_flags_byte);
}
@@ -1060,9 +1061,11 @@ impl Node {
return;
};
let now = std::time::Instant::now();
mmp.metrics
.process_receiver_report(&rr, our_timestamp_ms, now);
let (_first_rtt, rr_log) =
mmp.metrics
.process_receiver_report(&rr, our_timestamp_ms, crate::mmp::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);
// Feed SRTT back to sender/receiver report interval tuning (session-layer bounds)
if let Some(srtt_ms) = mmp.metrics.srtt_ms() {
@@ -1128,8 +1131,9 @@ impl Node {
};
let old_mtu = mmp.path_mtu.current_mtu();
let now = std::time::Instant::now();
let changed = mmp.path_mtu.apply_notification(notif.path_mtu, now);
let changed = mmp
.path_mtu
.apply_notification(notif.path_mtu, crate::mmp::mono_ms());
let new_mtu = mmp.path_mtu.current_mtu();
if !changed {
@@ -1342,8 +1346,10 @@ impl Node {
&& let Some(mmp) = entry.mmp_mut()
{
let old_mtu = mmp.path_mtu.current_mtu();
let now = std::time::Instant::now();
if mmp.path_mtu.apply_notification(msg.mtu, now) {
if mmp
.path_mtu
.apply_notification(msg.mtu, crate::mmp::mono_ms())
{
let new_mtu = mmp.path_mtu.current_mtu();
info!(
dest = %peer_name,
@@ -1928,7 +1934,7 @@ impl Node {
/// Similar to `send_session_data()` but:
/// - Takes an explicit `msg_type` byte (0x11, 0x12, 0x13, etc.)
/// - Never includes COORDS_PRESENT (reports are lightweight)
/// - Reads spin bit from MMP state for the inner header
/// - Reads the session timestamp for the inner header
/// - Records the send in MMP sender state
pub(in crate::node) async fn send_session_msg(
&mut self,
@@ -1938,7 +1944,7 @@ impl Node {
) -> Result<(), NodeError> {
let now_ms = Self::now_ms();
// Read spin bit and session timestamp from entry
// Read session timestamp from entry
let entry = self
.sessions
.get(dest_addr)
src/node/mod.rs
+9 -3
View File
@@ -60,6 +60,7 @@ use crate::peer::{ActivePeer, PeerConnection};
use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateLimiter};
use crate::proto::fmp::Fmp;
use crate::proto::fmp::NodeProfile;
use crate::proto::mmp::Mmp;
use crate::proto::routing::{self, Router, RoutingErrorRateLimiter};
#[cfg(unix)]
use crate::transport::ethernet::EthernetTransport;
@@ -440,6 +441,9 @@ pub struct Node {
/// FMP connection-lifecycle decision anchor (stateless; drives the
/// tick-poll maintain/teardown decisions).
fmp: Fmp,
/// MMP reporting decision anchor (stateless; drives the report-fan-out /
/// liveness / heartbeat decisions).
mmp: Mmp,
/// Rate limiter for source-side CoordsRequired/PathBroken responses.
coords_response_rate_limiter: RoutingErrorRateLimiter,
@@ -686,6 +690,7 @@ impl Node {
icmp_rate_limiter: IcmpRateLimiter::new(),
routing: Router::new(),
fmp: Fmp::new(),
mmp: Mmp::new(),
coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms(
coords_response_interval_ms,
),
@@ -848,6 +853,7 @@ impl Node {
icmp_rate_limiter: IcmpRateLimiter::new(),
routing: Router::new(),
fmp: Fmp::new(),
mmp: Mmp::new(),
coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms(
coords_response_interval_ms,
),
@@ -2079,7 +2085,7 @@ impl Node {
(Some(srtt), Some(setx)) => Some(setx * (1.0 + srtt / 100.0)),
_ => None,
};
let trend = |dual: &crate::mmp::algorithms::DualEwma| {
let trend = |dual: &crate::proto::mmp::DualEwma| {
dual.initialized()
.then(|| crate::control::queries::trend_label(dual.short(), dual.long()))
};
@@ -2119,7 +2125,7 @@ impl Node {
(Some(srtt), Some(setx)) => Some(setx * (1.0 + srtt / 100.0)),
_ => None,
};
let trend = |dual: &crate::mmp::algorithms::DualEwma| {
let trend = |dual: &crate::proto::mmp::DualEwma| {
dual.initialized()
.then(|| crate::control::queries::trend_label(dual.short(), dual.long()))
};
@@ -3074,7 +3080,7 @@ impl routing::RoutingView for NodeRoutingView<'_> {
/// is precomputed here exactly as the on-loop queries do, so the render is a
/// plain field emit.
fn project_entity_mmp(
metrics: &crate::mmp::metrics::MmpMetrics,
metrics: &crate::proto::mmp::MmpMetrics,
mode: String,
path_mtu: Option<u16>,
) -> crate::control::snapshot::EntityMmp {
src/node/session.rs
+9 -7
View File
@@ -5,13 +5,11 @@
//! (SessionSetup/SessionAck/SessionMsg3) carried inside SessionDatagram
//! envelopes through the mesh.
use std::time::Instant;
use crate::NodeAddr;
use crate::config::SessionMmpConfig;
use crate::mmp::MmpSessionState;
use crate::node::REKEY_JITTER_SECS;
use crate::noise::{HandshakeState, NoiseSession};
use crate::proto::mmp::MmpSessionState;
use rand::RngExt;
use secp256k1::PublicKey;
@@ -100,7 +98,7 @@ pub(crate) struct SessionEntry {
/// reset on CoordsRequired receipt.
coords_warmup_remaining: u8,
/// Whether this node initiated the Noise handshake.
/// Used for spin bit role assignment in session-layer MMP.
/// Surfaced through the `is_initiator()` accessor.
is_initiator: bool,
/// Session-layer MMP state. Initialized on Established transition.
mmp: Option<MmpSessionState>,
@@ -338,7 +336,11 @@ impl SessionEntry {
/// Initialize session-layer MMP state (called on Established transition).
pub(crate) fn init_mmp(&mut self, config: &SessionMmpConfig) {
self.mmp = Some(MmpSessionState::new(config, self.is_initiator));
self.mmp = Some(MmpSessionState::new(
config.mode,
config.log_interval_secs,
config.owd_window_size,
));
}
// === Traffic Counters ===
@@ -665,9 +667,9 @@ impl SessionEntry {
self.rekey_jitter_secs = draw_rekey_jitter();
// Reset MMP counters to avoid metric discontinuity
let now = Instant::now();
let now_ms = crate::mmp::mono_ms();
if let Some(mmp) = &mut self.mmp {
mmp.reset_for_rekey(now);
mmp.reset_for_rekey(now_ms);
}
true
}
src/node/tests/mmp_chartests.rs
+436
View File
@@ -0,0 +1,436 @@
//! Characterization tests for the three under-tested MMP tick handlers.
//!
//! These lock in the *current* observable behavior of the MMP fan-out and
//! first-RTT paths so a later behavior-neutral sans-IO extraction has an
//! equality oracle. The `check_link_heartbeats` handler already has a good
//! oracle (`heartbeat.rs` + `tcp.rs`) and is not re-covered here; this file
//! targets the three paths with no direct handler tests:
//!
//! * `check_mmp_reports` — link-layer mode/flag fan-out gating
//! * `check_session_mmp_reports` — session mode + PathMtu gating + backoff dedup
//! * `handle_receiver_report` — the first-RTT tree re-evaluation branch
//!
//! Assertions capture what the code does today, surprising or not.
//!
//! Report-generation is probed through the reused `src/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.
//!
//! Two `#[cfg(test)]` production seams are used, both on `ActivePeer`:
//! * `test_init_mmp(mode)` — attach link MMP with a chosen mode to a
//! bare (sessionless) peer, so mode gating is exercisable.
//! * `test_backdate_session_start` — age `session_elapsed_ms()` so a crafted
//! ReceiverReport yields a positive RTT sample (first-RTT trigger).
//!
//! Neither changes any decision logic or threshold.
use super::*;
use crate::config::SessionMmpConfig;
use crate::node::session::{EndToEndState, SessionEntry};
use crate::noise::HandshakeState;
use crate::peer::ActivePeer;
use crate::proto::mmp::{MmpMode, ReceiverReport};
use crate::tree::{ParentDeclaration, TreeCoordinate};
// ===========================================================================
// Helpers
// ===========================================================================
/// Insert a bare (sessionless) peer carrying link-layer MMP state in `mode`.
/// Returns the peer's NodeAddr.
fn insert_link_peer(node: &mut Node, mode: MmpMode) -> NodeAddr {
let identity = make_peer_identity();
let addr = *identity.node_addr();
let mut peer = ActivePeer::new(identity, LinkId::new(1), 0);
peer.test_init_mmp(mode);
node.peers.insert(addr, peer);
addr
}
/// Arm both sender and receiver link-MMP intervals so a report would be built.
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());
}
/// Complete an in-memory Noise XX handshake, returning the initiator session.
fn make_noise_session(
our_identity: &crate::Identity,
remote_identity: &crate::Identity,
) -> crate::noise::NoiseSession {
let mut initiator = HandshakeState::new_initiator(our_identity.keypair());
let mut responder = HandshakeState::new_responder(remote_identity.keypair());
let mut init_epoch = [0u8; 8];
rand::Rng::fill_bytes(&mut rand::rng(), &mut init_epoch);
initiator.set_local_epoch(init_epoch);
let mut resp_epoch = [0u8; 8];
rand::Rng::fill_bytes(&mut rand::rng(), &mut resp_epoch);
responder.set_local_epoch(resp_epoch);
let msg1 = initiator.write_message_1().unwrap();
responder.read_message_1(&msg1).unwrap();
let msg2 = responder.write_message_2().unwrap();
initiator.read_message_2(&msg2).unwrap();
let msg3 = initiator.write_message_3().unwrap();
responder.read_message_3(&msg3).unwrap();
initiator.into_session().unwrap()
}
/// Insert an Established session carrying session-layer MMP state in `mode`.
/// Returns the destination NodeAddr.
fn insert_session(node: &mut Node, mode: MmpMode) -> NodeAddr {
let remote = crate::Identity::generate();
let remote_addr = *remote.node_addr();
let session = make_noise_session(node.identity(), &remote);
let mut entry = SessionEntry::new(
remote_addr,
remote.pubkey_full(),
EndToEndState::Established(session),
1000,
true,
);
let cfg = SessionMmpConfig {
mode,
..SessionMmpConfig::default()
};
entry.init_mmp(&cfg);
node.sessions.insert(remote_addr, entry);
remote_addr
}
/// Arm both sender and receiver session-MMP intervals.
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());
}
// ===========================================================================
// check_mmp_reports — link-layer mode fan-out
// ===========================================================================
/// Full mode: both a SenderReport and a ReceiverReport are generated (both
/// intervals consumed).
#[tokio::test]
async fn mmp_full_mode_builds_sender_and_receiver_reports() {
let mut node = make_node();
let addr = insert_link_peer(&mut node, MmpMode::Full);
arm_link_mmp(&mut node, &addr);
node.check_mmp_reports().await;
let mmp = node.get_peer(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
assert!(
!mmp.sender.should_send_report(now),
"Full mode consumes the sender interval (SenderReport built)"
);
assert!(
!mmp.receiver.should_send_report(now),
"Full mode consumes the receiver interval (ReceiverReport built)"
);
}
/// Lightweight mode: only a ReceiverReport is generated; the sender interval
/// is left intact (no SenderReport in Lightweight).
#[tokio::test]
async fn mmp_lightweight_mode_builds_receiver_report_only() {
let mut node = make_node();
let addr = insert_link_peer(&mut node, MmpMode::Lightweight);
arm_link_mmp(&mut node, &addr);
node.check_mmp_reports().await;
let mmp = node.get_peer(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
assert!(
mmp.sender.should_send_report(now),
"Lightweight mode suppresses the SenderReport (sender interval intact)"
);
assert!(
!mmp.receiver.should_send_report(now),
"Lightweight mode still builds the ReceiverReport (receiver interval consumed)"
);
}
/// Minimal mode: neither report is generated; both intervals stay intact.
#[tokio::test]
async fn mmp_minimal_mode_builds_nothing() {
let mut node = make_node();
let addr = insert_link_peer(&mut node, MmpMode::Minimal);
arm_link_mmp(&mut node, &addr);
node.check_mmp_reports().await;
let mmp = node.get_peer(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
assert!(
mmp.sender.should_send_report(now),
"Minimal mode suppresses the SenderReport"
);
assert!(
mmp.receiver.should_send_report(now),
"Minimal mode suppresses the ReceiverReport"
);
}
/// Periodic operator logging fires once per interval: a fresh peer is due for
/// a log, and after one tick the log is marked (not due again within the
/// interval).
#[tokio::test]
async fn mmp_should_log_marks_logged_once_per_interval() {
let mut node = make_node();
let addr = insert_link_peer(&mut node, MmpMode::Full);
assert!(
node.get_peer(&addr)
.unwrap()
.mmp()
.unwrap()
.should_log(crate::mmp::mono_ms()),
"a freshly created peer is due for its first operator log"
);
node.check_mmp_reports().await;
assert!(
!node
.get_peer(&addr)
.unwrap()
.mmp()
.unwrap()
.should_log(crate::mmp::mono_ms()),
"after one tick the log is marked and not due again within the interval"
);
}
// ===========================================================================
// check_session_mmp_reports — session mode + PathMtu gating + backoff dedup
// ===========================================================================
/// Full mode session: both SenderReport and ReceiverReport are generated
/// (both intervals consumed) even though the send has no route and fails.
#[tokio::test]
async fn session_full_mode_builds_sender_and_receiver_reports() {
let mut node = make_node();
let addr = insert_session(&mut node, MmpMode::Full);
arm_session_mmp(&mut node, &addr);
node.check_session_mmp_reports().await;
let mmp = node.get_session(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
assert!(
!mmp.sender.should_send_report(now),
"Full session consumes the sender interval"
);
assert!(
!mmp.receiver.should_send_report(now),
"Full session consumes the receiver interval"
);
}
/// PathMtu notifications gate on all modes: in Minimal mode neither report is
/// built, yet a PathMtuNotification is still generated when an MTU has been
/// observed.
#[tokio::test]
async fn session_minimal_mode_still_sends_path_mtu() {
let mut node = make_node();
let addr = insert_session(&mut node, MmpMode::Minimal);
arm_session_mmp(&mut node, &addr);
// Observe an MTU so a notification becomes due (all modes).
node.sessions
.get_mut(&addr)
.unwrap()
.mmp_mut()
.unwrap()
.path_mtu
.observe_incoming_mtu(1200);
let now_before = crate::mmp::mono_ms();
assert!(
node.get_session(&addr)
.unwrap()
.mmp()
.unwrap()
.path_mtu
.should_send_notification(now_before),
"precondition: a PathMtuNotification is due after observing an MTU"
);
node.check_session_mmp_reports().await;
let mmp = node.get_session(&addr).unwrap().mmp().unwrap();
let now = crate::mmp::mono_ms();
assert!(
mmp.sender.should_send_report(now),
"Minimal mode suppresses the session SenderReport"
);
assert!(
mmp.receiver.should_send_report(now),
"Minimal mode suppresses the session ReceiverReport"
);
assert!(
!mmp.path_mtu.should_send_notification(now),
"PathMtuNotification is generated in Minimal mode (gate is mode-independent)"
);
}
/// Backoff dedup, all-fail side: a Full-mode session generates two reports
/// (SR + RR) to one destination; with no route both sends fail. The
/// per-destination dedup collapses the two failures into exactly ONE
/// `record_send_failure` (consecutive count advances by 1, not 2).
#[tokio::test]
async fn session_backoff_all_reports_fail_records_single_failure() {
let mut node = make_node();
let addr = insert_session(&mut node, MmpMode::Full);
arm_session_mmp(&mut node, &addr);
assert_eq!(
node.get_session(&addr)
.unwrap()
.mmp()
.unwrap()
.sender
.consecutive_send_failures(),
0,
"precondition: no prior send failures"
);
node.check_session_mmp_reports().await;
assert_eq!(
node.get_session(&addr)
.unwrap()
.mmp()
.unwrap()
.sender
.consecutive_send_failures(),
1,
"two failed reports to one dest dedup to a single record_send_failure"
);
}
// ===========================================================================
// handle_receiver_report — first-RTT tree re-evaluation branch
// ===========================================================================
/// Build a peer (NodeAddr strictly smaller than the node's own) that carries
/// link MMP but no RTT yet, and register it in the tree as a self-root with
/// that smaller address. This makes it a mandatory parent-switch target once
/// it becomes eligible. Returns the peer's NodeAddr.
fn setup_smaller_root_peer(node: &mut Node) -> NodeAddr {
let my_addr = *node.node_addr();
let (identity, addr) = loop {
let id = make_peer_identity();
let a = *id.node_addr();
if a < my_addr {
break (id, a);
}
};
let mut peer = ActivePeer::new(identity, LinkId::new(1), 0);
peer.test_init_mmp(MmpMode::Full);
// Age the session so a crafted ReceiverReport yields a positive RTT.
peer.test_backdate_session_start(std::time::Duration::from_secs(10));
node.peers.insert(addr, peer);
// Register the peer as a self-root in the tree at its (smaller) address.
node.tree_state_mut().update_peer(
ParentDeclaration::self_root(addr, 1, 0),
TreeCoordinate::root(addr),
);
addr
}
/// Craft a ReceiverReport whose timestamp echo yields a valid first RTT
/// sample. `highest`/`pkts`/`bytes` advance the cumulative counters so a
/// second report is not dropped as stale/duplicate.
fn craft_rr_payload(highest: u64, pkts: u64, bytes: u64) -> Vec<u8> {
let rr = ReceiverReport {
highest_counter: highest,
cumulative_packets_recv: pkts,
cumulative_bytes_recv: bytes,
timestamp_echo: 1000,
dwell_time: 0,
jitter: 0,
ecn_ce_count: 0,
owd_trend: 0,
burst_loss_count: 0,
cumulative_reorder_count: 0,
};
// handle_receiver_report receives the body with the msg_type byte stripped.
rr.encode()[1..].to_vec()
}
/// A first RTT sample flips the peer eligible for parent selection AND fires
/// the shell-resident tree branch: the node (initially self-root) adopts the
/// smaller-addressed peer as its new root.
#[tokio::test]
async fn first_rtt_flips_peer_eligible_and_triggers_tree_reeval() {
let mut node = make_node();
let addr = setup_smaller_root_peer(&mut node);
assert!(
node.tree_state().is_root(),
"precondition: node starts as its own root"
);
assert!(
!node.get_peer(&addr).unwrap().has_srtt(),
"precondition: peer has no RTT measurement yet"
);
let switches_before = node.metrics().tree.parent_switches.get();
node.handle_receiver_report(&addr, &craft_rr_payload(10, 5, 500))
.await;
assert!(
node.get_peer(&addr).unwrap().has_srtt(),
"first RTT sample makes the peer eligible for parent selection"
);
assert!(
!node.tree_state().is_root(),
"the first-RTT tree branch fired: node adopted a parent"
);
assert_eq!(
node.tree_state().root(),
&addr,
"node switched its root to the smaller-addressed peer"
);
assert!(
node.metrics().tree.parent_switches.get() > switches_before,
"the parent-switch was recorded in the tree metrics"
);
}
/// Regression guard: a *second* ReceiverReport (RTT already initialized, so
/// `first_rtt` is false) does NOT re-enter the tree branch — no further parent
/// switch is recorded.
#[tokio::test]
async fn non_first_receiver_report_does_not_retrigger_tree() {
let mut node = make_node();
let addr = setup_smaller_root_peer(&mut node);
// First report: fires the branch (established by the test above).
node.handle_receiver_report(&addr, &craft_rr_payload(10, 5, 500))
.await;
let switches_after_first = node.metrics().tree.parent_switches.get();
assert!(node.get_peer(&addr).unwrap().has_srtt());
// Second report with advanced counters: first_rtt is now false.
node.handle_receiver_report(&addr, &craft_rr_payload(20, 10, 1000))
.await;
assert_eq!(
node.metrics().tree.parent_switches.get(),
switches_after_first,
"a non-first ReceiverReport does not re-enter the first-RTT tree branch"
);
}
src/node/tests/mod.rs
+1
View File
@@ -19,6 +19,7 @@ mod ethernet;
mod forwarding;
mod handshake;
mod heartbeat;
mod mmp_chartests;
mod routing;
mod session;
mod spanning_tree;
src/node/tests/session.rs
+1 -1
View File
@@ -1908,7 +1908,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, std::time::Instant::now());
.apply_notification(reduced_mtu, crate::mmp::mono_ms());
assert_eq!(mmp.path_mtu.current_mtu(), reduced_mtu);
}