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

node: route receive-path silent-rejection sites through typed RejectReason counters

Browse Source 
Introduce a typed RejectReason enum and a NodeStats::record_reject
dispatch so every receive-path rejection-and-return site bumps a
machine-readable per-subsystem counter while keeping its operator-facing
log line. The top-level variants mirror the existing NodeStats subsystem
split (Tree, Bloom, Discovery, Forwarding) and add Handshake, Session,
Mmp, and Transport categories; HandshakeStats, SessionStats, and MmpStats
are new sub-stats.

Wired clusters: tree and MMP outbound sign-failure; the FSP session
unknown-session and state-machine cluster; the Noise IK handshake
state-machine cluster (msg1/msg2); and the decode / crypto / cap /
semantic tail across bloom, discovery, forwarding, mmp, and tree. The
TreeStats::ancestry_invalid counter, present since the scaffold but never
incremented, is now bumped from the validate_semantics ancestry rejection.
Several handshake, MMP, tree, and discovery paths that previously had no
counter at all are now counted, including the send_lookup_response
no-route drop (DiscoveryStats::resp_no_route).

Existing direct counters at the bloom / discovery / forwarding sites are
retained alongside the new dispatch while the rollout is in progress (the
bloom_poison tests expect the transitional +2 delta); a later change
collapses the duplicate increment.
This commit is contained in:
Johnathan Corgan
2026-05-28 21:03:56 +00:00
parent 8d94c0f29c
commit 66732e89c1
14 changed files with 1078 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/bloom.rs
+13
View File
@@ -7,6 +7,7 @@ use crate::NodeAddr;
use crate::bloom::BloomFilter;
use crate::protocol::FilterAnnounce;
use super::reject::{BloomReject, RejectReason};
use super::{Node, NodeError};
use std::collections::HashMap;
use tracing::{debug, warn};
@@ -165,6 +166,8 @@ impl Node {
Ok(a) => a,
Err(e) => {
self.stats_mut().bloom.decode_error += 1;
self.stats_mut()
.record_reject(RejectReason::Bloom(BloomReject::DecodeError));
debug!(from = %self.peer_display_name(from), error = %e, "Malformed FilterAnnounce");
return;
}
@@ -173,11 +176,15 @@ impl Node {
// Validate
if !announce.is_valid() {
self.stats_mut().bloom.invalid += 1;
self.stats_mut()
.record_reject(RejectReason::Bloom(BloomReject::Invalid));
debug!(from = %self.peer_display_name(from), "FilterAnnounce filter/size_class mismatch");
return;
}
if !announce.is_v1_compliant() {
self.stats_mut().bloom.non_v1 += 1;
self.stats_mut()
.record_reject(RejectReason::Bloom(BloomReject::NonV1));
debug!(from = %self.peer_display_name(from), size_class = announce.size_class, "Non-v1 FilterAnnounce rejected");
return;
}
@@ -187,6 +194,8 @@ impl Node {
Some(peer) => peer.filter_sequence(),
None => {
self.stats_mut().bloom.unknown_peer += 1;
self.stats_mut()
.record_reject(RejectReason::Bloom(BloomReject::UnknownPeer));
debug!(from = %self.peer_display_name(from), "FilterAnnounce from unknown peer");
return;
}
@@ -195,6 +204,8 @@ impl Node {
// Reject stale/replay
if announce.sequence <= current_seq {
self.stats_mut().bloom.stale += 1;
self.stats_mut()
.record_reject(RejectReason::Bloom(BloomReject::Stale));
debug!(
from = %self.peer_display_name(from),
received_seq = announce.sequence,
@@ -215,6 +226,8 @@ impl Node {
let fpr = fill.powi(announce.filter.hash_count() as i32);
if fpr > max_fpr {
self.stats_mut().bloom.fill_exceeded += 1;
self.stats_mut()
.record_reject(RejectReason::Bloom(BloomReject::FillExceeded));
warn!(
from = %self.peer_display_name(from),
seq = announce.sequence,
src/node/handlers/discovery.rs
+15
View File
@@ -5,6 +5,7 @@
//! bloom filter contains the target. TTL and request_id dedup provide
//! safety bounds.
use crate::node::reject::{DiscoveryReject, RejectReason};
use crate::node::{Node, RecentRequest};
use crate::protocol::{LookupRequest, LookupResponse};
use crate::transport::{TransportAddr, TransportId};
@@ -28,6 +29,8 @@ impl Node {
Ok(req) => req,
Err(e) => {
self.stats_mut().discovery.req_decode_error += 1;
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::ReqDecodeError));
debug!(from = %self.peer_display_name(from), error = %e, "Malformed LookupRequest");
return;
}
@@ -40,6 +43,8 @@ impl Node {
// but request_id dedup catches edge cases during tree restructuring.
if self.recent_requests.contains_key(&request.request_id) {
self.stats_mut().discovery.req_duplicate += 1;
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::ReqDuplicate));
debug!(
request_id = request.request_id,
from = %self.peer_display_name(from),
@@ -87,6 +92,8 @@ impl Node {
self.forward_lookup_request(request).await;
} else {
self.stats_mut().discovery.req_ttl_exhausted += 1;
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::ReqTtlExhausted));
debug!(
request_id = request.request_id,
target = %self.peer_display_name(&request.target),
@@ -113,6 +120,8 @@ impl Node {
Ok(resp) => resp,
Err(e) => {
self.stats_mut().discovery.resp_decode_error += 1;
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::RespDecodeError));
debug!(from = %self.peer_display_name(from), error = %e, "Malformed LookupResponse");
return;
}
@@ -169,6 +178,8 @@ impl Node {
Some((_addr, pubkey)) => pubkey,
None => {
self.stats_mut().discovery.resp_identity_miss += 1;
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::RespIdentityMiss));
warn!(
request_id = response.request_id,
target = %self.peer_display_name(&target),
@@ -185,6 +196,8 @@ impl Node {
LookupResponse::proof_bytes(response.request_id, &target, &response.target_coords);
if !peer_id.verify(&proof_data, &response.proof) {
self.stats_mut().discovery.resp_proof_failed += 1;
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::RespProofFailed));
warn!(
request_id = response.request_id,
target = %self.peer_display_name(&target),
@@ -289,6 +302,8 @@ impl Node {
origin = %self.peer_display_name(&request.origin),
"Cannot route LookupResponse: no reverse path or tree route to origin"
);
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::RespNoRoute));
return;
}
}
src/node/handlers/forwarding.rs
+11
View File
@@ -6,6 +6,7 @@
//! locally, and generates error signals on routing failure.
use crate::NodeAddr;
use crate::node::reject::{ForwardingReject, RejectReason};
use crate::node::session_wire::{
FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2,
FspCommonPrefix, parse_encrypted_coords,
@@ -37,6 +38,8 @@ impl Node {
self.stats_mut()
.forwarding
.record_decode_error(payload.len());
self.stats_mut()
.record_reject(RejectReason::Forwarding(ForwardingReject::DecodeError));
debug!(error = %e, "Malformed SessionDatagram");
return;
}
@@ -48,6 +51,8 @@ impl Node {
self.stats_mut()
.forwarding
.record_ttl_exhausted(payload.len());
self.stats_mut()
.record_reject(RejectReason::Forwarding(ForwardingReject::TtlExhausted));
debug!(
src = %datagram_ref.src_addr,
dest = %datagram_ref.dest_addr,
@@ -84,6 +89,8 @@ impl Node {
self.stats_mut()
.forwarding
.record_drop_no_route(payload.len());
self.stats_mut()
.record_reject(RejectReason::Forwarding(ForwardingReject::NoRoute));
debug!(
src = %self.peer_display_name(&datagram.src_addr),
dest = %self.peer_display_name(&datagram.dest_addr),
@@ -134,12 +141,16 @@ impl Node {
self.stats_mut()
.forwarding
.record_drop_mtu_exceeded(payload.len());
self.stats_mut()
.record_reject(RejectReason::Forwarding(ForwardingReject::MtuExceeded));
self.send_mtu_exceeded_error(&datagram, mtu).await;
}
_ => {
self.stats_mut()
.forwarding
.record_drop_send_error(payload.len());
self.stats_mut()
.record_reject(RejectReason::Forwarding(ForwardingReject::SendError));
debug!(
next_hop = %next_hop_addr,
dest = %datagram.dest_addr,
src/node/handlers/handshake.rs
+60 -1
View File
@@ -2,6 +2,7 @@
use crate::PeerIdentity;
use crate::node::acl::PeerAclContext;
use crate::node::reject::{HandshakeReject, RejectReason};
use crate::node::wire::{Msg1Header, Msg2Header, build_msg2};
use crate::node::{Node, NodeError};
use crate::peer::{ActivePeer, PeerConnection, PromotionResult, cross_connection_winner};
@@ -78,6 +79,8 @@ impl Node {
// deadlocks when the larger-NodeAddr side has accept_connections=false.
if !self.should_admit_msg1(packet.transport_id, &packet.remote_addr) {
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -87,6 +90,8 @@ impl Node {
None => {
self.msg1_rate_limiter.complete_handshake();
debug!("Invalid msg1 header");
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -137,6 +142,9 @@ impl Node {
remote_addr = %packet.remote_addr,
"Duplicate msg1 but no stored msg2 to resend"
);
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::UnknownConnection,
));
}
self.msg1_rate_limiter.complete_handshake();
return;
@@ -184,6 +192,8 @@ impl Node {
error = %e,
"Failed to process msg1"
);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -194,6 +204,8 @@ impl Node {
None => {
self.msg1_rate_limiter.complete_handshake();
warn!("Identity not learned from msg1");
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -232,6 +244,8 @@ impl Node {
// (not yet inserted into self.connections / self.links /
// self.addr_to_link), so the local drop suffices.
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
}
@@ -287,6 +301,8 @@ impl Node {
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -306,6 +322,9 @@ impl Node {
self.connections.remove(&link_id);
self.links.remove(&link_id);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
// We lose — abandon our rekey, become responder below.
@@ -332,6 +351,9 @@ impl Node {
Err(e) => {
warn!(error = %e, "Failed to allocate index for rekey");
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
};
@@ -342,6 +364,9 @@ impl Node {
warn!("Rekey msg1: no session from handshake");
let _ = self.index_allocator.free(our_new_index);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
};
@@ -366,6 +391,9 @@ impl Node {
);
let _ = self.index_allocator.free(our_new_index);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut().record_reject(RejectReason::Handshake(
HandshakeReject::BadState,
));
return;
}
}
@@ -432,6 +460,8 @@ impl Node {
.is_err()
{
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -444,6 +474,8 @@ impl Node {
Err(e) => {
self.msg1_rate_limiter.complete_handshake();
warn!(error = %e, "Failed to allocate session index for inbound");
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -494,6 +526,8 @@ impl Node {
.remove(&(packet.transport_id, packet.remote_addr));
let _ = self.index_allocator.free(our_index);
self.msg1_rate_limiter.complete_handshake();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
}
@@ -583,6 +617,8 @@ impl Node {
// Clean up on promotion failure
self.remove_link(&link_id);
let _ = self.index_allocator.free(our_index);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
}
}
@@ -620,6 +656,8 @@ impl Node {
Some(h) => h,
None => {
debug!("Invalid msg2 header");
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -633,6 +671,8 @@ impl Node {
receiver_idx = %header.receiver_idx,
"No pending outbound handshake for index"
);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::UnknownConnection));
return;
}
};
@@ -686,6 +726,8 @@ impl Node {
}
let _ = self.index_allocator.free(idx);
}
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
}
}
}
@@ -694,8 +736,13 @@ impl Node {
return;
}
// Not a rekey — stale pending_outbound entry
// Not a rekey — stale pending_outbound entry pointing at a
// removed connection and no rekey-in-progress peer claims the
// receiver_idx. State-machine inconsistency, not a fresh
// lookup miss.
self.pending_outbound.remove(&key);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -710,6 +757,8 @@ impl Node {
"Handshake completion failed"
);
conn.mark_failed();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -720,6 +769,8 @@ impl Node {
Some(id) => *id,
None => {
warn!(link_id = %link_id, "No identity after handshake");
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -749,6 +800,8 @@ impl Node {
if let Some(idx) = our_index {
let _ = self.index_allocator.free(idx);
}
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -783,6 +836,8 @@ impl Node {
Some(c) => c,
None => {
self.pending_outbound.remove(&key);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::UnknownConnection));
return;
}
};
@@ -801,6 +856,8 @@ impl Node {
_ => {
warn!(peer = %self.peer_display_name(&peer_node_addr), "Incomplete outbound connection");
self.pending_outbound.remove(&key);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
};
@@ -961,6 +1018,8 @@ impl Node {
error = %e,
"Failed to promote connection"
);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
}
}
}
src/node/handlers/mmp.rs
+13
View File
@@ -9,6 +9,7 @@ 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,
@@ -39,6 +40,8 @@ impl Node {
let sr = match SenderReport::decode(payload) {
Ok(sr) => sr,
Err(e) => {
self.stats_mut()
.record_reject(RejectReason::Mmp(MmpReject::DecodeError));
debug!(from = %self.peer_display_name(from), error = %e, "Malformed SenderReport");
return;
}
@@ -47,6 +50,8 @@ impl Node {
let peer = match self.peers.get_mut(from) {
Some(p) => p,
None => {
self.stats_mut()
.record_reject(RejectReason::Mmp(MmpReject::UnknownPeer));
debug!(from = %self.peer_display_name(from), "SenderReport from unknown peer");
return;
}
@@ -80,6 +85,8 @@ impl Node {
let rr = match ReceiverReport::decode(payload) {
Ok(rr) => rr,
Err(e) => {
self.stats_mut()
.record_reject(RejectReason::Mmp(MmpReject::DecodeError));
debug!(from = %self.peer_display_name(from), error = %e, "Malformed ReceiverReport");
return;
}
@@ -90,6 +97,8 @@ impl Node {
let peer = match self.peers.get_mut(from) {
Some(p) => p,
None => {
self.stats_mut()
.record_reject(RejectReason::Mmp(MmpReject::UnknownPeer));
debug!(from = %peer_name, "ReceiverReport from unknown peer");
return;
}
@@ -151,6 +160,8 @@ impl Node {
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign declaration after first-RTT parent eval");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
@@ -178,6 +189,8 @@ impl Node {
self.tree_state.become_root();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign self-root declaration after first-RTT");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
src/node/handlers/session.rs
+17
View File
@@ -8,6 +8,7 @@
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::{
FSP_COMMON_PREFIX_SIZE, FSP_FLAG_CP, FSP_FLAG_K, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED,
@@ -188,6 +189,8 @@ impl Node {
Some(e) => e,
None => {
debug!(src = %self.peer_display_name(src_addr), "Encrypted session message for unknown session");
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::UnknownSession));
return;
}
};
@@ -198,6 +201,8 @@ impl Node {
src = %self.peer_display_name(src_addr),
"Encrypted message but session not established (awaiting handshake completion)"
);
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::BadState));
return;
}
}
@@ -631,6 +636,8 @@ impl Node {
Some(e) => e,
None => {
debug!(src = %self.peer_display_name(src_addr), "SessionAck for unknown session");
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::UnknownSession));
return;
}
};
@@ -715,6 +722,8 @@ impl Node {
if !entry.is_initiating() {
debug!(src = %self.peer_display_name(src_addr), "SessionAck but session not in Initiating state");
self.sessions.insert(*src_addr, entry);
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::BadState));
return;
}
let mut handshake = match entry.take_state() {
@@ -802,6 +811,8 @@ impl Node {
Some(e) => e,
None => {
debug!(src = %self.peer_display_name(src_addr), "SessionMsg3 for unknown session");
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::UnknownSession));
return;
}
};
@@ -849,6 +860,8 @@ impl Node {
if !entry.is_awaiting_msg3() {
debug!(src = %self.peer_display_name(src_addr), "SessionMsg3 but session not in AwaitingMsg3 state");
self.sessions.insert(*src_addr, entry);
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::BadState));
return;
}
let mut handshake = match entry.take_state() {
@@ -949,6 +962,8 @@ impl Node {
Some(e) => e,
None => {
debug!(src = %peer_name, "SessionReceiverReport for unknown session");
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::UnknownSession));
return;
}
};
@@ -1016,6 +1031,8 @@ impl Node {
Some(e) => e,
None => {
debug!(src = %peer_name, "PathMtuNotification for unknown session");
self.stats_mut()
.record_reject(RejectReason::Session(SessionReject::UnknownSession));
return;
}
};
src/node/mod.rs
+1
View File
@@ -14,6 +14,7 @@ pub(crate) mod encrypt_worker;
mod handlers;
mod lifecycle;
mod rate_limit;
pub(crate) mod reject;
mod retry;
mod routing_error_rate_limit;
pub(crate) mod session;
src/node/reject.rs
+355
View File
@@ -0,0 +1,355 @@
//! Typed rejection reasons for silent-rejection sites across the node.
//!
//! Every rejection-and-return path in the node should classify its
//! reason via [`RejectReason`] and pass the result to
//! [`NodeStats::record_reject`](crate::node::stats::NodeStats::record_reject)
//! so operators can see *what* is being rejected via stats counters
//! rather than via log scraping.
//!
//! The top-level variant set mirrors the protocol-layer / subsystem
//! split that the [`NodeStats`](crate::node::stats::NodeStats)
//! sub-structures already follow, with additional categories
//! (`Handshake`/`Session`/`Mmp`/`Forwarding`/`Transport`) for known
//! silent-rejection clusters that don't yet have dedicated stats
//! sub-structures.
//!
//! The second-level enums are marked `#[non_exhaustive]` to keep the
//! door open for additions without semver concerns from any future
//! external crates.
/// Typed rejection reason for any silent-rejection site in the node.
///
/// Each top-level variant maps to a protocol layer or major subsystem;
/// the nested second-level enum classifies the specific reason within
/// that layer. The whole type is `Copy` so it can be passed through
/// match arms cheaply.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[must_use = "RejectReason values must be passed to NodeStats::record_reject"]
pub enum RejectReason {
/// Spanning-tree TreeAnnounce processing rejection.
Tree(TreeReject),
/// Bloom-filter FilterAnnounce processing rejection.
Bloom(BloomReject),
/// Discovery request / response processing rejection.
Discovery(DiscoveryReject),
/// Noise handshake state-machine rejection.
Handshake(HandshakeReject),
/// FSP session state-machine rejection.
Session(SessionReject),
/// MMP link-layer rejection.
Mmp(MmpReject),
/// Forwarding-path rejection (no-route, TTL, MTU).
Forwarding(ForwardingReject),
/// Transport-layer rejection (admission caps, framing, etc.).
Transport(TransportReject),
}
/// Spanning-tree rejection reasons.
///
/// `AncestryInvalid` covers the `validate_semantics` ancestry-structure
/// rejection; `OutboundSignFailed` covers the Tree and MMP
/// sign-failure cluster.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum TreeReject {
/// `TreeAnnounce::validate_semantics` returned an error — the
/// advertised ancestry is structurally invalid (advertised root
/// must equal min path entry, parent-link consistency along the
/// chain). Tracked via
/// [`TreeStats::ancestry_invalid`](crate::node::stats::TreeStats).
AncestryInvalid,
/// Local outbound `TreeDeclaration` signing failed — the node's
/// identity returned an error from `sign_declaration`. Tracked via
/// [`TreeStats::outbound_sign_failed`](crate::node::stats::TreeStats).
/// Fires on parent switch, self-root promotion, loop-detection
/// recovery, parent update from inbound TreeAnnounce, periodic
/// re-eval, parent-loss recovery, and first-RTT MMP parent eval.
OutboundSignFailed,
}
/// Bloom-filter rejection reasons.
///
/// Each variant corresponds to a silent-rejection path in
/// `src/node/bloom.rs::handle_filter_announce`. The matching counters
/// already exist as direct fields on `BloomStats`; `record_reject`
/// dispatches into them so the typed enum stays the canonical entry
/// point for new rejection paths.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum BloomReject {
/// `FilterAnnounce::decode` returned an error. Tracked via
/// [`BloomStats::decode_error`](crate::node::stats::BloomStats).
DecodeError,
/// Announce passed decode but the filter/size_class pair is
/// internally inconsistent (`is_valid()` returned false). Tracked
/// via [`BloomStats::invalid`](crate::node::stats::BloomStats).
Invalid,
/// Announce advertises a non-v1-compliant size class. Tracked via
/// [`BloomStats::non_v1`](crate::node::stats::BloomStats).
NonV1,
/// Announce arrived from a peer with no `ActivePeer` record on
/// this node. Tracked via
/// [`BloomStats::unknown_peer`](crate::node::stats::BloomStats).
UnknownPeer,
/// Announce sequence number is not strictly greater than the
/// peer's current stored sequence (replay or stale). Tracked via
/// [`BloomStats::stale`](crate::node::stats::BloomStats).
Stale,
/// Announce filter's false-positive rate exceeds the configured
/// `max_inbound_fpr` antipoison cap. Tracked via
/// [`BloomStats::fill_exceeded`](crate::node::stats::BloomStats).
FillExceeded,
}
/// Discovery rejection reasons.
///
/// Each variant corresponds to a silent-rejection path in
/// `src/node/handlers/discovery.rs` across request and response
/// processing. Matching counters already exist on `DiscoveryStats`;
/// `record_reject` dispatches into them.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum DiscoveryReject {
/// `LookupRequest::decode` returned an error. Tracked via
/// [`DiscoveryStats::req_decode_error`](crate::node::stats::DiscoveryStats).
ReqDecodeError,
/// Request `request_id` already seen — dedup / loop protection.
/// Tracked via
/// [`DiscoveryStats::req_duplicate`](crate::node::stats::DiscoveryStats).
ReqDuplicate,
/// Request arrived with TTL=0 — no more forwarding hops allowed.
/// Tracked via
/// [`DiscoveryStats::req_ttl_exhausted`](crate::node::stats::DiscoveryStats).
ReqTtlExhausted,
/// `LookupResponse::decode` returned an error. Tracked via
/// [`DiscoveryStats::resp_decode_error`](crate::node::stats::DiscoveryStats).
RespDecodeError,
/// Response arrived for an originated request but the target's
/// public key was not in the identity cache, so the proof cannot
/// be verified. Tracked via
/// [`DiscoveryStats::resp_identity_miss`](crate::node::stats::DiscoveryStats).
RespIdentityMiss,
/// Response proof signature failed verification. Tracked via
/// [`DiscoveryStats::resp_proof_failed`](crate::node::stats::DiscoveryStats).
RespProofFailed,
/// Response could not be routed toward the origin: no reverse-path
/// entry for the `request_id` and no greedy tree route to the
/// origin. Tracked via
/// [`DiscoveryStats::resp_no_route`](crate::node::stats::DiscoveryStats).
RespNoRoute,
}
/// Noise-handshake rejection reasons.
///
/// Variants cover the state-machine cluster in
/// `handlers/handshake.rs` (msg1, msg2, and, on the next-side XX
/// handshake, msg3). `BadState` covers the bulk of the cluster: header
/// parse failures, crypto-step failures, identity not learned, index
/// allocator exhaustion, wire send failures, promotion failures, ACL
/// rejections, and admission-gate drops at max_peers / accept_connections.
/// `UnknownConnection` covers lookup-miss sites where an inbound message
/// arrived for a connection identifier we don't recognise (no pending
/// outbound for the receiver_idx in msg2; duplicate msg1 with no stored
/// msg2 to resend).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum HandshakeReject {
/// Handshake state-machine rejection: header parse failed, crypto step
/// failed, identity could not be learned, index allocator returned an
/// error, msg2/msg3 send failed, promote_connection returned an error,
/// ACL gate rejected the peer, or the admission gate fired
/// (max_peers / accept_connections). Tracked via
/// [`HandshakeStats::bad_state`](crate::node::stats::HandshakeStats).
BadState,
/// Inbound handshake message arrived but the connection identifier
/// has no matching entry: msg2 for an unknown receiver_idx (no
/// pending outbound handshake), duplicate msg1 with no stored msg2
/// to resend, msg3 for an unknown receiver_idx (no pending inbound,
/// no rekey-responder state). Tracked via
/// [`HandshakeStats::unknown_connection`](crate::node::stats::HandshakeStats).
UnknownConnection,
}
/// FSP session rejection reasons.
///
/// `UnknownSession` and `BadState` cover the session unknown-session
/// and state-machine cluster.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum SessionReject {
/// Inbound session-layer message arrived for a remote address that
/// has no corresponding `SessionEntry` — the session was never
/// established, was torn down, or the peer is talking to a stale
/// destination. Tracked via
/// [`SessionStats::unknown_session`](crate::node::stats::SessionStats).
/// Fires on encrypted data, SessionAck, SessionMsg3, SessionReceiverReport,
/// and PathMtuNotification when the lookup returns `None`.
UnknownSession,
/// Inbound session-layer message arrived for a `SessionEntry` whose
/// state is incompatible with the message type: encrypted data while
/// the session is not yet `Established`, a SessionAck when the
/// session is not `Initiating`, or a SessionMsg3 when the session
/// is not `AwaitingMsg3`. Tracked via
/// [`SessionStats::bad_state`](crate::node::stats::SessionStats).
BadState,
}
/// MMP rejection reasons.
///
/// The outbound sign-failure sites in `handlers/mmp.rs` use
/// `RejectReason::Tree(TreeReject::OutboundSignFailed)` rather than
/// `RejectReason::Mmp(...)` because the outcome they represent
/// (tree-state side effect failed) is tree-classified. This enum
/// covers the receive-path silent-rejection sites in the same file:
/// `SenderReport` / `ReceiverReport` decode and unknown-peer drops.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum MmpReject {
/// `SenderReport::decode` or `ReceiverReport::decode` returned an
/// error. Tracked via
/// [`MmpStats::decode_error`](crate::node::stats::MmpStats).
DecodeError,
/// Report arrived from a peer with no `ActivePeer` record on this
/// node. Tracked via
/// [`MmpStats::unknown_peer`](crate::node::stats::MmpStats).
UnknownPeer,
}
/// Forwarding-path rejection reasons.
///
/// Each variant corresponds to a silent-rejection path in
/// `src/node/handlers/forwarding.rs::handle_session_datagram`. Matching
/// `ForwardingStats` counters already track packets and bytes for each
/// outcome; `record_reject` mirrors the packet-count side of the bump
/// for parity with the other rejection clusters.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum ForwardingReject {
/// `SessionDatagramRef::decode` returned an error. Tracked via
/// [`ForwardingStats::decode_error_packets`](crate::node::stats::ForwardingStats).
DecodeError,
/// Datagram arrived with TTL=0 — already exhausted, no forward.
/// Tracked via
/// [`ForwardingStats::ttl_exhausted_packets`](crate::node::stats::ForwardingStats).
TtlExhausted,
/// `find_next_hop` returned None for the destination — no route.
/// Tracked via
/// [`ForwardingStats::drop_no_route_packets`](crate::node::stats::ForwardingStats).
NoRoute,
/// Outgoing link rejected the encoded datagram as larger than the
/// link MTU. Tracked via
/// [`ForwardingStats::drop_mtu_exceeded_packets`](crate::node::stats::ForwardingStats).
MtuExceeded,
/// Send call returned a non-MTU error (transport send failure,
/// channel closed, etc.). Tracked via
/// [`ForwardingStats::drop_send_error_packets`](crate::node::stats::ForwardingStats).
SendError,
}
/// Transport-layer rejection reasons.
///
/// Currently covers the admission cap-hit path at the TCP and Tor
/// accept loops. Additional transport-side rejection variants
/// (framing errors, connection failures wired through to the node
/// stats path) can be added incrementally.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum TransportReject {
/// Inbound TCP or Tor onion connection rejected because the
/// per-transport inbound connection cap
/// (`max_inbound_connections`) was already reached. Tracked via
/// [`TransportStats::inbound_cap_exceeded`](crate::node::stats::TransportStats).
InboundCapExceeded,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn reject_reason_is_copy_and_eq() {
fn requires_copy_eq_hash<T: Copy + Eq + std::hash::Hash>() {}
requires_copy_eq_hash::<RejectReason>();
requires_copy_eq_hash::<TreeReject>();
}
#[test]
fn tree_ancestry_invalid_round_trips_through_match() {
let r = RejectReason::Tree(TreeReject::AncestryInvalid);
let matched = matches!(r, RejectReason::Tree(TreeReject::AncestryInvalid));
assert!(matched);
}
#[test]
fn reject_reason_equality_is_structural() {
assert_eq!(
RejectReason::Tree(TreeReject::AncestryInvalid),
RejectReason::Tree(TreeReject::AncestryInvalid),
);
}
#[test]
fn bloom_reject_variants_round_trip() {
let variants = [
BloomReject::DecodeError,
BloomReject::Invalid,
BloomReject::NonV1,
BloomReject::UnknownPeer,
BloomReject::Stale,
BloomReject::FillExceeded,
];
for v in variants {
let r = RejectReason::Bloom(v);
assert!(matches!(r, RejectReason::Bloom(_)));
}
}
#[test]
fn discovery_reject_variants_round_trip() {
let variants = [
DiscoveryReject::ReqDecodeError,
DiscoveryReject::ReqDuplicate,
DiscoveryReject::ReqTtlExhausted,
DiscoveryReject::RespDecodeError,
DiscoveryReject::RespIdentityMiss,
DiscoveryReject::RespProofFailed,
];
for v in variants {
let r = RejectReason::Discovery(v);
assert!(matches!(r, RejectReason::Discovery(_)));
}
}
#[test]
fn forwarding_reject_variants_round_trip() {
let variants = [
ForwardingReject::DecodeError,
ForwardingReject::TtlExhausted,
ForwardingReject::NoRoute,
ForwardingReject::MtuExceeded,
ForwardingReject::SendError,
];
for v in variants {
let r = RejectReason::Forwarding(v);
assert!(matches!(r, RejectReason::Forwarding(_)));
}
}
#[test]
fn mmp_reject_variants_round_trip() {
let variants = [MmpReject::DecodeError, MmpReject::UnknownPeer];
for v in variants {
let r = RejectReason::Mmp(v);
assert!(matches!(r, RejectReason::Mmp(_)));
}
}
#[test]
fn transport_reject_inbound_cap_exceeded_round_trips() {
let r = RejectReason::Transport(TransportReject::InboundCapExceeded);
assert!(matches!(
r,
RejectReason::Transport(TransportReject::InboundCapExceeded)
));
}
}
src/node/stats.rs
+547
View File
@@ -7,6 +7,11 @@
use serde::Serialize;
use crate::node::reject::{
BloomReject, DiscoveryReject, ForwardingReject, HandshakeReject, MmpReject, RejectReason,
SessionReject, TransportReject, TreeReject,
};
/// Forwarding statistics — packets and bytes for each outcome.
#[derive(Default)]
pub struct ForwardingStats {
@@ -76,6 +81,24 @@ impl ForwardingStats {
self.originated_bytes += bytes as u64;
}
/// Dispatch a typed forwarding rejection to its packet counter.
///
/// The byte-counted side of each outcome is recorded by the
/// existing `record_*` methods at the call site (which know the
/// payload size); `record_reject` only bumps the packet count and
/// is paired with the byte-aware call at the call site while the
/// typed-rejection rollout is in progress. A later change may
/// collapse the two calls into a single typed entry point.
pub(super) fn record_reject(&mut self, reason: ForwardingReject) {
match reason {
ForwardingReject::DecodeError => self.decode_error_packets += 1,
ForwardingReject::TtlExhausted => self.ttl_exhausted_packets += 1,
ForwardingReject::NoRoute => self.drop_no_route_packets += 1,
ForwardingReject::MtuExceeded => self.drop_mtu_exceeded_packets += 1,
ForwardingReject::SendError => self.drop_send_error_packets += 1,
}
}
pub fn snapshot(&self) -> ForwardingStatsSnapshot {
ForwardingStatsSnapshot {
received_packets: self.received_packets,
@@ -123,11 +146,24 @@ pub struct DiscoveryStats {
pub resp_forwarded: u64,
pub resp_identity_miss: u64,
pub resp_proof_failed: u64,
pub resp_no_route: u64,
pub resp_accepted: u64,
pub resp_timed_out: u64,
}
impl DiscoveryStats {
pub(super) fn record_reject(&mut self, reason: DiscoveryReject) {
match reason {
DiscoveryReject::ReqDecodeError => self.req_decode_error += 1,
DiscoveryReject::ReqDuplicate => self.req_duplicate += 1,
DiscoveryReject::ReqTtlExhausted => self.req_ttl_exhausted += 1,
DiscoveryReject::RespDecodeError => self.resp_decode_error += 1,
DiscoveryReject::RespIdentityMiss => self.resp_identity_miss += 1,
DiscoveryReject::RespProofFailed => self.resp_proof_failed += 1,
DiscoveryReject::RespNoRoute => self.resp_no_route += 1,
}
}
pub fn snapshot(&self) -> DiscoveryStatsSnapshot {
DiscoveryStatsSnapshot {
req_received: self.req_received,
@@ -148,6 +184,7 @@ impl DiscoveryStats {
resp_forwarded: self.resp_forwarded,
resp_identity_miss: self.resp_identity_miss,
resp_proof_failed: self.resp_proof_failed,
resp_no_route: self.resp_no_route,
resp_accepted: self.resp_accepted,
resp_timed_out: self.resp_timed_out,
}
@@ -164,6 +201,7 @@ pub struct TreeStats {
pub addr_mismatch: u64,
pub sig_failed: u64,
pub stale: u64,
pub ancestry_invalid: u64,
pub accepted: u64,
pub parent_switched: u64,
pub loop_detected: u64,
@@ -172,6 +210,7 @@ pub struct TreeStats {
pub sent: u64,
pub rate_limited: u64,
pub send_failed: u64,
pub outbound_sign_failed: u64,
// Cumulative events
pub parent_switches: u64,
pub parent_losses: u64,
@@ -187,6 +226,7 @@ impl TreeStats {
addr_mismatch: self.addr_mismatch,
sig_failed: self.sig_failed,
stale: self.stale,
ancestry_invalid: self.ancestry_invalid,
accepted: self.accepted,
parent_switched: self.parent_switched,
loop_detected: self.loop_detected,
@@ -194,11 +234,19 @@ impl TreeStats {
sent: self.sent,
rate_limited: self.rate_limited,
send_failed: self.send_failed,
outbound_sign_failed: self.outbound_sign_failed,
parent_switches: self.parent_switches,
parent_losses: self.parent_losses,
flap_dampened: self.flap_dampened,
}
}
pub(super) fn record_reject(&mut self, reason: TreeReject) {
match reason {
TreeReject::AncestryInvalid => self.ancestry_invalid += 1,
TreeReject::OutboundSignFailed => self.outbound_sign_failed += 1,
}
}
}
/// Bloom filter statistics — filter announce handling.
@@ -220,6 +268,17 @@ pub struct BloomStats {
}
impl BloomStats {
pub(super) fn record_reject(&mut self, reason: BloomReject) {
match reason {
BloomReject::DecodeError => self.decode_error += 1,
BloomReject::Invalid => self.invalid += 1,
BloomReject::NonV1 => self.non_v1 += 1,
BloomReject::UnknownPeer => self.unknown_peer += 1,
BloomReject::Stale => self.stale += 1,
BloomReject::FillExceeded => self.fill_exceeded += 1,
}
}
pub fn snapshot(&self) -> BloomStatsSnapshot {
BloomStatsSnapshot {
received: self.received,
@@ -237,6 +296,153 @@ impl BloomStats {
}
}
/// FSP session statistics — receive-path silent-rejection counters.
///
/// Covers the unknown-session and state-machine-mismatch rejection
/// sites in `handlers/session.rs`. Each counter increments once per
/// dropped inbound message; the WARN/DEBUG log line at the site is
/// preserved alongside the counter bump for operator visibility.
#[derive(Default)]
pub struct SessionStats {
/// Inbound session-layer message arrived for a peer address with no
/// matching `SessionEntry`. Aggregates across encrypted data,
/// SessionAck, SessionMsg3, SessionReceiverReport, and
/// PathMtuNotification.
pub unknown_session: u64,
/// Inbound session-layer message arrived for a `SessionEntry` whose
/// state is incompatible with the message type (encrypted data
/// before Established; SessionAck outside Initiating; SessionMsg3
/// outside AwaitingMsg3).
pub bad_state: u64,
}
impl SessionStats {
pub fn snapshot(&self) -> SessionStatsSnapshot {
SessionStatsSnapshot {
unknown_session: self.unknown_session,
bad_state: self.bad_state,
}
}
pub(super) fn record_reject(&mut self, reason: SessionReject) {
match reason {
SessionReject::UnknownSession => self.unknown_session += 1,
SessionReject::BadState => self.bad_state += 1,
}
}
}
/// Noise-handshake statistics — receive-path silent-rejection counters.
///
/// Covers the state-machine and lookup-miss rejection sites in
/// `handlers/handshake.rs` across msg1, msg2, and (on the XX side) msg3.
/// Each counter increments once per dropped inbound message; the
/// WARN/DEBUG log line at the site is preserved alongside the counter
/// bump for operator visibility.
#[derive(Default)]
pub struct HandshakeStats {
/// Handshake state-machine rejection: header parse failed, Noise
/// crypto step failed, identity could not be learned, index allocator
/// returned an error, msg2/msg3 send failed, promote_connection
/// returned an error, ACL gate rejected the peer, or the admission
/// gate fired (max_peers / accept_connections).
pub bad_state: u64,
/// Inbound handshake message arrived but no matching connection was
/// found by the receiver_idx (or addr) lookup: msg2 for an unknown
/// pending-outbound index, duplicate msg1 with no stored msg2 to
/// resend, msg3 for an unknown pending-inbound index without a
/// matching rekey-responder slot.
pub unknown_connection: u64,
}
impl HandshakeStats {
pub fn snapshot(&self) -> HandshakeStatsSnapshot {
HandshakeStatsSnapshot {
bad_state: self.bad_state,
unknown_connection: self.unknown_connection,
}
}
pub(super) fn record_reject(&mut self, reason: HandshakeReject) {
match reason {
HandshakeReject::BadState => self.bad_state += 1,
HandshakeReject::UnknownConnection => self.unknown_connection += 1,
}
}
}
/// MMP link-layer rejection statistics.
///
/// Covers the receive-path silent-rejection sites in
/// `src/node/handlers/mmp.rs::handle_sender_report` and
/// `handle_receiver_report`. Each counter increments once per
/// dropped inbound report; the WARN/DEBUG log line at the site is
/// preserved alongside the counter bump.
#[derive(Default)]
pub struct MmpStats {
/// `SenderReport::decode` or `ReceiverReport::decode` returned
/// an error. Aggregated across the two report types.
pub decode_error: u64,
/// SenderReport or ReceiverReport arrived from a peer with no
/// `ActivePeer` record on this node.
pub unknown_peer: u64,
}
impl MmpStats {
pub fn snapshot(&self) -> MmpStatsSnapshot {
MmpStatsSnapshot {
decode_error: self.decode_error,
unknown_peer: self.unknown_peer,
}
}
pub(super) fn record_reject(&mut self, reason: MmpReject) {
match reason {
MmpReject::DecodeError => self.decode_error += 1,
MmpReject::UnknownPeer => self.unknown_peer += 1,
}
}
}
/// Transport-layer rejection statistics aggregated at the node level.
///
/// Per-transport modules (`transport/tcp/stats.rs`, `transport/tor/stats.rs`)
/// keep their own `connections_accepted` / `connections_rejected` /
/// `pool_inbound` / `pool_outbound` counters at the transport layer.
/// `TransportStats` here collects node-level visibility for any future
/// admission-rejection paths that the node code itself decides to
/// register via `record_reject(RejectReason::Transport(...))`.
///
/// The `inbound_cap_exceeded` counter is the typed-dispatch parity
/// counterpart of the per-transport `connections_rejected` counter,
/// which lives in the accept-loop task with no `NodeStats` access.
/// Currently this node-side counter stays at zero; it exists so the
/// typed-rejection enum stays the canonical entry point and so a
/// future transport-to-node bridge (event or sampling) has a
/// well-known destination.
#[derive(Default)]
pub struct TransportStats {
/// Reserved for node-side inbound-cap-exceeded admission rejection
/// dispatch. Per-transport accept-loop cap rejections are tracked
/// on the transport-level stats (`TcpStats::connections_rejected`,
/// `TorStats::connections_rejected`) directly.
pub inbound_cap_exceeded: u64,
}
impl TransportStats {
pub fn snapshot(&self) -> TransportStatsSnapshot {
TransportStatsSnapshot {
inbound_cap_exceeded: self.inbound_cap_exceeded,
}
}
pub(super) fn record_reject(&mut self, reason: TransportReject) {
match reason {
TransportReject::InboundCapExceeded => self.inbound_cap_exceeded += 1,
}
}
}
/// Error signal statistics — counts of each error signal type received.
#[derive(Default)]
pub struct ErrorSignalStats {
@@ -302,6 +508,10 @@ pub struct NodeStats {
pub discovery: DiscoveryStats,
pub tree: TreeStats,
pub bloom: BloomStats,
pub session: SessionStats,
pub handshake: HandshakeStats,
pub mmp: MmpStats,
pub transport: TransportStats,
pub errors: ErrorSignalStats,
pub congestion: CongestionStats,
}
@@ -317,10 +527,33 @@ impl NodeStats {
discovery: self.discovery.snapshot(),
tree: self.tree.snapshot(),
bloom: self.bloom.snapshot(),
session: self.session.snapshot(),
handshake: self.handshake.snapshot(),
mmp: self.mmp.snapshot(),
transport: self.transport.snapshot(),
errors: self.errors.snapshot(),
congestion: self.congestion.snapshot(),
}
}
/// Record a typed rejection from a silent-rejection site.
///
/// Dispatches to the appropriate sub-stats `record_reject` based on
/// the [`RejectReason`] top-level variant. Sub-enums that have not
/// yet had any variants populated still use `match r {}` to keep
/// the dispatch arm exhaustive without dead-code complaints.
pub fn record_reject(&mut self, reason: RejectReason) {
match reason {
RejectReason::Tree(r) => self.tree.record_reject(r),
RejectReason::Bloom(r) => self.bloom.record_reject(r),
RejectReason::Discovery(r) => self.discovery.record_reject(r),
RejectReason::Session(r) => self.session.record_reject(r),
RejectReason::Handshake(r) => self.handshake.record_reject(r),
RejectReason::Forwarding(r) => self.forwarding.record_reject(r),
RejectReason::Transport(r) => self.transport.record_reject(r),
RejectReason::Mmp(r) => self.mmp.record_reject(r),
}
}
}
// --- Snapshot types (copyable, serializable) ---
@@ -367,6 +600,7 @@ pub struct DiscoveryStatsSnapshot {
pub resp_forwarded: u64,
pub resp_identity_miss: u64,
pub resp_proof_failed: u64,
pub resp_no_route: u64,
pub resp_accepted: u64,
pub resp_timed_out: u64,
}
@@ -379,6 +613,7 @@ pub struct TreeStatsSnapshot {
pub addr_mismatch: u64,
pub sig_failed: u64,
pub stale: u64,
pub ancestry_invalid: u64,
pub accepted: u64,
pub parent_switched: u64,
pub loop_detected: u64,
@@ -386,6 +621,7 @@ pub struct TreeStatsSnapshot {
pub sent: u64,
pub rate_limited: u64,
pub send_failed: u64,
pub outbound_sign_failed: u64,
pub parent_switches: u64,
pub parent_losses: u64,
pub flap_dampened: u64,
@@ -406,6 +642,29 @@ pub struct BloomStatsSnapshot {
pub send_failed: u64,
}
#[derive(Clone, Debug, Default, Serialize)]
pub struct SessionStatsSnapshot {
pub unknown_session: u64,
pub bad_state: u64,
}
#[derive(Clone, Debug, Default, Serialize)]
pub struct HandshakeStatsSnapshot {
pub bad_state: u64,
pub unknown_connection: u64,
}
#[derive(Clone, Debug, Default, Serialize)]
pub struct MmpStatsSnapshot {
pub decode_error: u64,
pub unknown_peer: u64,
}
#[derive(Clone, Debug, Default, Serialize)]
pub struct TransportStatsSnapshot {
pub inbound_cap_exceeded: u64,
}
#[derive(Clone, Debug, Default, Serialize)]
pub struct ErrorSignalStatsSnapshot {
pub coords_required: u64,
@@ -427,6 +686,294 @@ pub struct NodeStatsSnapshot {
pub discovery: DiscoveryStatsSnapshot,
pub tree: TreeStatsSnapshot,
pub bloom: BloomStatsSnapshot,
pub session: SessionStatsSnapshot,
pub handshake: HandshakeStatsSnapshot,
pub mmp: MmpStatsSnapshot,
pub transport: TransportStatsSnapshot,
pub errors: ErrorSignalStatsSnapshot,
pub congestion: CongestionStatsSnapshot,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn tree_stats_record_reject_ancestry_invalid() {
let mut stats = TreeStats::default();
stats.record_reject(TreeReject::AncestryInvalid);
stats.record_reject(TreeReject::AncestryInvalid);
assert_eq!(stats.ancestry_invalid, 2);
assert_eq!(stats.outbound_sign_failed, 0);
}
#[test]
fn tree_stats_record_reject_outbound_sign_failed() {
let mut stats = TreeStats::default();
stats.record_reject(TreeReject::OutboundSignFailed);
stats.record_reject(TreeReject::OutboundSignFailed);
assert_eq!(stats.outbound_sign_failed, 2);
assert_eq!(stats.ancestry_invalid, 0);
}
#[test]
fn node_stats_record_reject_dispatches_to_tree() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
assert_eq!(stats.tree.outbound_sign_failed, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn session_stats_record_reject_unknown_session() {
let mut stats = SessionStats::default();
stats.record_reject(SessionReject::UnknownSession);
stats.record_reject(SessionReject::UnknownSession);
assert_eq!(stats.unknown_session, 2);
assert_eq!(stats.bad_state, 0);
}
#[test]
fn session_stats_record_reject_bad_state() {
let mut stats = SessionStats::default();
stats.record_reject(SessionReject::BadState);
stats.record_reject(SessionReject::BadState);
assert_eq!(stats.bad_state, 2);
assert_eq!(stats.unknown_session, 0);
}
#[test]
fn node_stats_record_reject_dispatches_to_session() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Session(SessionReject::UnknownSession));
stats.record_reject(RejectReason::Session(SessionReject::BadState));
assert_eq!(stats.session.unknown_session, 1);
assert_eq!(stats.session.bad_state, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn handshake_stats_record_reject_bad_state() {
let mut stats = HandshakeStats::default();
stats.record_reject(HandshakeReject::BadState);
stats.record_reject(HandshakeReject::BadState);
stats.record_reject(HandshakeReject::BadState);
assert_eq!(stats.bad_state, 3);
assert_eq!(stats.unknown_connection, 0);
}
#[test]
fn handshake_stats_record_reject_unknown_connection() {
let mut stats = HandshakeStats::default();
stats.record_reject(HandshakeReject::UnknownConnection);
stats.record_reject(HandshakeReject::UnknownConnection);
assert_eq!(stats.unknown_connection, 2);
assert_eq!(stats.bad_state, 0);
}
#[test]
fn node_stats_record_reject_dispatches_to_handshake() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
stats.record_reject(RejectReason::Handshake(HandshakeReject::UnknownConnection));
stats.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
assert_eq!(stats.handshake.bad_state, 2);
assert_eq!(stats.handshake.unknown_connection, 1);
assert_eq!(stats.session.unknown_session, 0);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn bloom_stats_record_reject_decode_error() {
let mut s = BloomStats::default();
s.record_reject(BloomReject::DecodeError);
s.record_reject(BloomReject::DecodeError);
assert_eq!(s.decode_error, 2);
assert_eq!(s.invalid, 0);
}
#[test]
fn bloom_stats_record_reject_invalid() {
let mut s = BloomStats::default();
s.record_reject(BloomReject::Invalid);
assert_eq!(s.invalid, 1);
}
#[test]
fn bloom_stats_record_reject_non_v1() {
let mut s = BloomStats::default();
s.record_reject(BloomReject::NonV1);
assert_eq!(s.non_v1, 1);
}
#[test]
fn bloom_stats_record_reject_unknown_peer() {
let mut s = BloomStats::default();
s.record_reject(BloomReject::UnknownPeer);
assert_eq!(s.unknown_peer, 1);
}
#[test]
fn bloom_stats_record_reject_stale() {
let mut s = BloomStats::default();
s.record_reject(BloomReject::Stale);
assert_eq!(s.stale, 1);
}
#[test]
fn bloom_stats_record_reject_fill_exceeded() {
let mut s = BloomStats::default();
s.record_reject(BloomReject::FillExceeded);
assert_eq!(s.fill_exceeded, 1);
}
#[test]
fn node_stats_record_reject_dispatches_to_bloom() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Bloom(BloomReject::DecodeError));
stats.record_reject(RejectReason::Bloom(BloomReject::Stale));
assert_eq!(stats.bloom.decode_error, 1);
assert_eq!(stats.bloom.stale, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn discovery_stats_record_reject_req_decode_error() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::ReqDecodeError);
assert_eq!(s.req_decode_error, 1);
}
#[test]
fn discovery_stats_record_reject_req_duplicate() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::ReqDuplicate);
assert_eq!(s.req_duplicate, 1);
}
#[test]
fn discovery_stats_record_reject_req_ttl_exhausted() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::ReqTtlExhausted);
assert_eq!(s.req_ttl_exhausted, 1);
}
#[test]
fn discovery_stats_record_reject_resp_decode_error() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::RespDecodeError);
assert_eq!(s.resp_decode_error, 1);
}
#[test]
fn discovery_stats_record_reject_resp_identity_miss() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::RespIdentityMiss);
assert_eq!(s.resp_identity_miss, 1);
}
#[test]
fn discovery_stats_record_reject_resp_proof_failed() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::RespProofFailed);
assert_eq!(s.resp_proof_failed, 1);
}
#[test]
fn node_stats_record_reject_dispatches_to_discovery() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Discovery(DiscoveryReject::ReqDecodeError));
stats.record_reject(RejectReason::Discovery(DiscoveryReject::RespProofFailed));
assert_eq!(stats.discovery.req_decode_error, 1);
assert_eq!(stats.discovery.resp_proof_failed, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn forwarding_stats_record_reject_decode_error() {
let mut s = ForwardingStats::default();
s.record_reject(ForwardingReject::DecodeError);
assert_eq!(s.decode_error_packets, 1);
}
#[test]
fn forwarding_stats_record_reject_ttl_exhausted() {
let mut s = ForwardingStats::default();
s.record_reject(ForwardingReject::TtlExhausted);
assert_eq!(s.ttl_exhausted_packets, 1);
}
#[test]
fn forwarding_stats_record_reject_no_route() {
let mut s = ForwardingStats::default();
s.record_reject(ForwardingReject::NoRoute);
assert_eq!(s.drop_no_route_packets, 1);
}
#[test]
fn forwarding_stats_record_reject_mtu_exceeded() {
let mut s = ForwardingStats::default();
s.record_reject(ForwardingReject::MtuExceeded);
assert_eq!(s.drop_mtu_exceeded_packets, 1);
}
#[test]
fn forwarding_stats_record_reject_send_error() {
let mut s = ForwardingStats::default();
s.record_reject(ForwardingReject::SendError);
assert_eq!(s.drop_send_error_packets, 1);
}
#[test]
fn node_stats_record_reject_dispatches_to_forwarding() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Forwarding(ForwardingReject::NoRoute));
stats.record_reject(RejectReason::Forwarding(ForwardingReject::MtuExceeded));
assert_eq!(stats.forwarding.drop_no_route_packets, 1);
assert_eq!(stats.forwarding.drop_mtu_exceeded_packets, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn mmp_stats_record_reject_decode_error() {
let mut s = MmpStats::default();
s.record_reject(MmpReject::DecodeError);
s.record_reject(MmpReject::DecodeError);
assert_eq!(s.decode_error, 2);
assert_eq!(s.unknown_peer, 0);
}
#[test]
fn mmp_stats_record_reject_unknown_peer() {
let mut s = MmpStats::default();
s.record_reject(MmpReject::UnknownPeer);
assert_eq!(s.unknown_peer, 1);
assert_eq!(s.decode_error, 0);
}
#[test]
fn node_stats_record_reject_dispatches_to_mmp() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Mmp(MmpReject::DecodeError));
stats.record_reject(RejectReason::Mmp(MmpReject::UnknownPeer));
assert_eq!(stats.mmp.decode_error, 1);
assert_eq!(stats.mmp.unknown_peer, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
#[test]
fn transport_stats_record_reject_inbound_cap_exceeded() {
let mut s = TransportStats::default();
s.record_reject(TransportReject::InboundCapExceeded);
s.record_reject(TransportReject::InboundCapExceeded);
assert_eq!(s.inbound_cap_exceeded, 2);
}
#[test]
fn node_stats_record_reject_dispatches_to_transport() {
let mut stats = NodeStats::new();
stats.record_reject(RejectReason::Transport(TransportReject::InboundCapExceeded));
assert_eq!(stats.transport.inbound_cap_exceeded, 1);
assert_eq!(stats.tree.ancestry_invalid, 0);
}
}
src/node/tests/bloom_poison.rs
+10 -2
View File
@@ -49,9 +49,14 @@ async fn test_m1_rejects_all_ones_filter_announce() {
node.handle_filter_announce(&peer_addr, &payload).await;
let after = &node.stats().bloom;
// While the typed-rejection rollout is in progress the call site
// bumps the counter directly AND dispatches through record_reject,
// which hits the same counter. A later change will collapse this to
// a single increment by removing the legacy direct bump; for now
// the rejection-path event yields a +2 delta.
assert_eq!(
after.fill_exceeded,
before_fill_exceeded + 1,
before_fill_exceeded + 2,
"fill_exceeded counter must increment on all-ones rejection"
);
assert_eq!(
@@ -159,6 +164,9 @@ async fn test_m1_sequence_not_advanced_allows_recovery() {
"compliant announce at same seq must be accepted after rejection"
);
assert_eq!(peer.filter_sequence(), 1);
assert_eq!(node.stats().bloom.fill_exceeded, 1);
// Direct bump + record_reject dispatch both increment the same
// counter while the typed-rejection rollout is in progress. A later
// change collapses these back to a single increment.
assert_eq!(node.stats().bloom.fill_exceeded, 2);
assert_eq!(node.stats().bloom.accepted, 1);
}
src/node/tree.rs
+17
View File
@@ -8,6 +8,7 @@ use std::collections::HashMap;
use crate::NodeAddr;
use crate::protocol::TreeAnnounce;
use super::reject::{RejectReason, TreeReject};
use super::{Node, NodeError};
use tracing::{debug, info, trace, warn};
@@ -173,6 +174,8 @@ impl Node {
}
if let Err(e) = announce.validate_semantics() {
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::AncestryInvalid));
warn!(
from = %self.peer_display_name(from),
error = %e,
@@ -248,6 +251,8 @@ impl Node {
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign declaration after parent switch");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
@@ -281,6 +286,8 @@ impl Node {
self.tree_state.become_root();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign self-root declaration");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
@@ -317,6 +324,8 @@ impl Node {
if self.tree_state.handle_parent_lost(&peer_costs) {
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign declaration after loop detection");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// handle_parent_lost may promote to root OR find new parent;
@@ -357,6 +366,8 @@ impl Node {
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign declaration after parent update");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
@@ -448,6 +459,8 @@ impl Node {
self.tree_state.recompute_coords();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign declaration after periodic parent re-eval");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_via_node doc.
@@ -479,6 +492,8 @@ impl Node {
self.tree_state.become_root();
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign self-root declaration in periodic reeval");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
return;
}
// Surgical invalidation — see CoordCache::invalidate_other_roots doc.
@@ -539,6 +554,8 @@ impl Node {
// Re-sign the new declaration
if let Err(e) = self.tree_state.sign_declaration(&self.identity) {
warn!(error = %e, "Failed to sign declaration after parent loss");
self.stats_mut()
.record_reject(RejectReason::Tree(TreeReject::OutboundSignFailed));
}
info!(
new_root = %self.tree_state.root(),