Merge receive-path reject-reason discipline and reloadable config consolidation

Bring the refactor-hotpath integration branch into master: explicit
RejectReason counters for previously-silent receive-path drop sites
across the tree, discovery, and handshake handlers, plus the Reloadable
trait and ArcSwap-backed hot-reload consolidation for the host map and
peer ACL. Includes the new discovery dedup-cache-full reject counter.
This commit is contained in:
Johnathan Corgan
2026-05-30 01:50:57 +00:00
20 changed files with 1546 additions and 59 deletions
+16
View File
@@ -9,6 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Typed `RejectReason` classification for receive-path silent-rejection
sites across the node. Each rejection-and-return path now passes a
typed reason to `NodeStats::record_reject`, which routes it to a
per-subsystem counter, so operators can see what is being rejected
through stats counters rather than by scraping debug logs. New
`HandshakeStats`, `SessionStats`, and `MmpStats` sub-stats join the
existing `TreeStats`, `BloomStats`, `DiscoveryStats`, and
`ForwardingStats`, and `TreeStats::ancestry_invalid` is now
incremented from the `TreeAnnounce::validate_semantics` rejection
site that was previously silent. Several handshake, MMP, tree, and
discovery rejection paths that 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;
a later change collapses the duplicate increment.
- `Node::update_peers` for runtime peer-list refresh, returning an
`UpdatePeersOutcome` summarizing added, removed, and retained peers.
Re-derives active peer connections from a new peer configuration
Generated
+10
View File
@@ -98,6 +98,15 @@ version = "1.0.102"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
[[package]]
name = "arc-swap"
version = "1.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207"
dependencies = [
"rustversion",
]
[[package]]
name = "arrayvec"
version = "0.7.6"
@@ -1063,6 +1072,7 @@ checksum = "9844ddc3a6e533d62bba727eb6c28b5d360921d5175e9ff0f1e621a5c590a4d5"
name = "fips"
version = "0.4.0-dev"
dependencies = [
"arc-swap",
"bech32",
"bluer",
"clap",
+1
View File
@@ -36,6 +36,7 @@ portable-atomic = { version = "1", features = ["std"] }
nostr = { version = "0.44", features = ["std", "nip59"] }
nostr-sdk = "0.44"
arc-swap = "1"
[target.'cfg(unix)'.dependencies]
tun = { version = "0.8.7", features = ["async"] }
+2
View File
@@ -11,6 +11,7 @@
"req_backoff_suppressed": 0,
"req_bloom_miss": 0,
"req_decode_error": 0,
"req_dedup_cache_full": 0,
"req_deduplicated": 0,
"req_duplicate": 0,
"req_fallback_forwarded": 0,
@@ -25,6 +26,7 @@
"resp_decode_error": 0,
"resp_forwarded": 0,
"resp_identity_miss": 0,
"resp_no_route": 0,
"resp_proof_failed": 0,
"resp_received": 0,
"resp_timed_out": 0
+2
View File
@@ -17,9 +17,11 @@
"accepted": 0,
"addr_mismatch": 0,
"ancestry_changed": 0,
"ancestry_invalid": 0,
"decode_error": 0,
"flap_dampened": 0,
"loop_detected": 0,
"outbound_sign_failed": 0,
"parent_losses": 0,
"parent_switched": 0,
"parent_switches": 0,
+55 -36
View File
@@ -9,6 +9,7 @@
//! evaluated first, an allowlist match overrides a denylist match for the
//! same peer.
use crate::node::reloadable::Reloadable;
use crate::node::{Node, NodeError};
use crate::transport::{TransportAddr, TransportId};
use crate::upper::hosts::{DEFAULT_HOSTS_PATH, HostMap, HostMapReloader, file_mtime};
@@ -17,6 +18,7 @@ use serde::Serialize;
use std::collections::{BTreeSet, HashSet};
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::SystemTime;
use tracing::{debug, info, warn};
@@ -283,8 +285,16 @@ impl PeerAcl {
}
/// Tracks peer ACL files and reloads them on mtime changes.
///
/// Follows the canonical Arc-wrapper template from [`Reloadable`]: the
/// reader-facing [`PeerAcl`] snapshot is published through an
/// [`arc_swap::ArcSwap`] so the authorization hot path reads it without
/// locking, while the reloader's change-detection state (file mtimes, the
/// embedded hosts reloader) is touched only by [`Reloadable::reload`] on the
/// single node tick task.
pub struct PeerAclReloader {
acl: PeerAcl,
/// Reader-facing effective ACL snapshot.
acl: arc_swap::ArcSwap<PeerAcl>,
hosts: HostMapReloader,
allow_path: PathBuf,
deny_path: PathBuf,
@@ -328,7 +338,7 @@ impl PeerAclReloader {
let acl = PeerAcl::load_files_with_hosts(&allow_path, &deny_path, hosts.hosts());
Self {
acl,
acl: arc_swap::ArcSwap::from(Arc::new(acl)),
hosts,
allow_path,
deny_path,
@@ -337,30 +347,34 @@ impl PeerAclReloader {
}
}
/// Get the current ACL.
pub fn acl(&self) -> &PeerAcl {
&self.acl
/// Acquire a lock-free guard over the current ACL snapshot.
pub fn acl(&self) -> arc_swap::Guard<Arc<PeerAcl>> {
self.load()
}
/// Return a human-readable snapshot of the loaded ACL state.
pub fn status(&self) -> PeerAclStatus {
let acl = self.acl.load();
PeerAclStatus {
allow_file: self.allow_path.display().to_string(),
deny_file: self.deny_path.display().to_string(),
enforcement_active: !self.acl.is_empty(),
effective_mode: self.acl.effective_mode().to_string(),
default_decision: self.acl.default_decision().to_string(),
allow_all: self.acl.allow_all,
deny_all: self.acl.deny_all,
allow_file_entries: self.acl.allow_file_entries(),
deny_file_entries: self.acl.deny_file_entries(),
allow_entries: self.acl.allow_entries(),
deny_entries: self.acl.deny_entries(),
enforcement_active: !acl.is_empty(),
effective_mode: acl.effective_mode().to_string(),
default_decision: acl.default_decision().to_string(),
allow_all: acl.allow_all,
deny_all: acl.deny_all,
allow_file_entries: acl.allow_file_entries(),
deny_file_entries: acl.deny_file_entries(),
allow_entries: acl.allow_entries(),
deny_entries: acl.deny_entries(),
}
}
}
/// Check whether ACL or hosts alias sources changed and reload if needed.
pub fn check_reload(&mut self) -> bool {
impl Reloadable for PeerAclReloader {
type Snapshot = PeerAcl;
async fn reload(&mut self) -> bool {
let allow_mtime = file_mtime(&self.allow_path);
let deny_mtime = file_mtime(&self.deny_path);
let hosts_changed = self.hosts.check_reload();
@@ -374,27 +388,32 @@ impl PeerAclReloader {
self.last_allow_mtime = allow_mtime;
self.last_deny_mtime = deny_mtime;
self.acl =
let new_acl =
PeerAcl::load_files_with_hosts(&self.allow_path, &self.deny_path, self.hosts.hosts());
info!(
allow_file = %self.allow_path.display(),
deny_file = %self.deny_path.display(),
allow_entries = self.acl.allow.len(),
deny_entries = self.acl.deny.len(),
allow_entries = new_acl.allow.len(),
deny_entries = new_acl.deny.len(),
alias_entries = self.hosts.hosts().len(),
allow_all = self.acl.allow_all,
deny_all = self.acl.deny_all,
allow_all = new_acl.allow_all,
deny_all = new_acl.deny_all,
"Reloaded peer ACL files"
);
self.acl.store(Arc::new(new_acl));
true
}
fn load(&self) -> arc_swap::Guard<Arc<PeerAcl>> {
self.acl.load()
}
}
impl Node {
/// Reload the peer ACL if the ACL or hosts files changed.
pub(crate) fn reload_peer_acl(&mut self) -> bool {
self.peer_acl.check_reload()
pub(crate) async fn reload_peer_acl(&mut self) -> bool {
self.peer_acl.reload().await
}
/// Return a control-plane snapshot of the current peer ACL.
@@ -747,20 +766,20 @@ mod tests {
assert_eq!(acl.check(&test_peer(&npub)), PeerAclDecision::AllowList);
}
#[test]
fn test_acl_reloader_detects_change() {
#[tokio::test]
async fn test_acl_reloader_detects_change() {
let dir = tempfile::tempdir().unwrap();
let allow = dir.path().join("peers.allow");
let deny = dir.path().join("peers.deny");
let denied = test_npub();
let mut reloader = PeerAclReloader::with_paths(allow.clone(), deny.clone());
assert!(!reloader.check_reload());
assert!(!reloader.reload().await);
std::thread::sleep(std::time::Duration::from_millis(5));
std::fs::write(&deny, format!("{denied}\n")).unwrap();
assert!(reloader.check_reload());
assert!(reloader.reload().await);
assert_eq!(
reloader
.acl()
@@ -769,8 +788,8 @@ mod tests {
);
}
#[test]
fn test_acl_reloader_detects_allow_file_removal() {
#[tokio::test]
async fn test_acl_reloader_detects_allow_file_removal() {
let dir = tempfile::tempdir().unwrap();
let allow = dir.path().join("peers.allow");
let deny = dir.path().join("peers.deny");
@@ -786,7 +805,7 @@ mod tests {
std::thread::sleep(std::time::Duration::from_millis(5));
std::fs::remove_file(&allow).unwrap();
assert!(reloader.check_reload());
assert!(reloader.reload().await);
assert!(reloader.acl().is_empty());
assert_eq!(
reloader.acl().check(&test_peer(&allowed)),
@@ -892,8 +911,8 @@ mod tests {
assert_eq!(acl.check(&peer), PeerAclDecision::AllowList);
}
#[test]
fn test_acl_reloader_detects_hosts_change_for_alias_entry() {
#[tokio::test]
async fn test_acl_reloader_detects_hosts_change_for_alias_entry() {
let dir = tempfile::tempdir().unwrap();
let allow = dir.path().join("peers.allow");
let deny = dir.path().join("peers.deny");
@@ -909,7 +928,7 @@ mod tests {
std::thread::sleep(std::time::Duration::from_millis(5));
std::fs::write(&hosts, format!("node-a {npub}\n")).unwrap();
assert!(reloader.check_reload());
assert!(reloader.reload().await);
assert_eq!(
reloader.acl().allow_file_entries(),
vec!["node-a".to_string()]
@@ -923,8 +942,8 @@ mod tests {
);
}
#[test]
fn test_acl_reloader_detects_hosts_removal_for_alias_entry() {
#[tokio::test]
async fn test_acl_reloader_detects_hosts_removal_for_alias_entry() {
let dir = tempfile::tempdir().unwrap();
let allow = dir.path().join("peers.allow");
let deny = dir.path().join("peers.deny");
@@ -944,7 +963,7 @@ mod tests {
std::thread::sleep(std::time::Duration::from_millis(5));
std::fs::remove_file(&hosts).unwrap();
assert!(reloader.check_reload());
assert!(reloader.reload().await);
assert!(reloader.acl().is_empty());
assert_eq!(
reloader.acl().check(&test_peer(&npub)),
+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,
+17
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};
@@ -30,6 +31,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;
}
@@ -43,6 +46,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),
@@ -52,6 +57,8 @@ impl Node {
}
if self.recent_requests.len() >= MAX_RECENT_DISCOVERY_REQUESTS {
self.stats_mut()
.record_reject(RejectReason::Discovery(DiscoveryReject::ReqDedupCacheFull));
debug!(
request_id = request.request_id,
from = %self.peer_display_name(from),
@@ -98,6 +105,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),
@@ -124,6 +133,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;
}
@@ -180,6 +191,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),
@@ -196,6 +209,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),
@@ -300,6 +315,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;
}
}
+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,
+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;
}
};
@@ -706,6 +746,8 @@ impl Node {
}
let _ = self.index_allocator.free(idx);
}
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
}
}
}
@@ -714,8 +756,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;
}
@@ -730,6 +777,8 @@ impl Node {
"Handshake completion failed"
);
conn.mark_failed();
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
@@ -740,6 +789,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;
}
};
@@ -769,6 +820,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;
}
@@ -803,6 +856,8 @@ impl Node {
Some(c) => c,
None => {
self.pending_outbound.remove(&key);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::UnknownConnection));
return;
}
};
@@ -821,6 +876,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;
}
};
@@ -981,6 +1038,8 @@ impl Node {
error = %e,
"Failed to promote connection"
);
self.stats_mut()
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
}
}
}
+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.
+9 -1
View File
@@ -249,7 +249,15 @@ impl Node {
_ = tick.tick() => {
self.check_timeouts();
let now_ms = Self::now_ms();
self.reload_peer_acl();
self.reload_peer_acl().await;
// The host map hot-reloads on the same tick as the ACL. It
// is polled separately from `reload_peer_acl` because the
// ACL's embedded alias reloader and this snapshot are
// distinct resources; the `path_mtu_lookup` cache and the
// `nostr_discovery` subsystem are deliberately excluded
// from `Reloadable` since neither reloads from a backing
// file (see `node::reloadable`).
self.reload_host_map().await;
self.poll_pending_connects().await;
self.poll_nostr_discovery().await;
self.poll_lan_discovery().await;
+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;
}
};
+21 -15
View File
@@ -14,6 +14,8 @@ pub(crate) mod encrypt_worker;
mod handlers;
mod lifecycle;
mod rate_limit;
pub(crate) mod reject;
mod reloadable;
mod retry;
mod routing_error_rate_limit;
pub(crate) mod session;
@@ -27,6 +29,7 @@ pub(crate) mod wire;
use self::discovery_rate_limit::{DiscoveryBackoff, DiscoveryForwardRateLimiter};
use self::rate_limit::HandshakeRateLimiter;
use self::reloadable::Reloadable;
use self::routing_error_rate_limit::RoutingErrorRateLimiter;
/// Half-range of the symmetric jitter applied to the per-session rekey timer.
@@ -518,8 +521,9 @@ pub struct Node {
// === Host Map ===
/// Static hostname → npub mapping for DNS resolution.
/// Built at construction from peer aliases and /etc/fips/hosts.
host_map: Arc<HostMap>,
/// Built at construction from peer aliases and /etc/fips/hosts, and
/// published through a lock-free snapshot for the display path.
host_map: reloadable::HostMapReloadable,
/// Off-task FMP-encrypt + UDP-send worker pool. Unix-only —
/// the worker issues direct sendmmsg(2) / sendmsg+UDP_GSO calls
@@ -612,13 +616,9 @@ impl Node {
let forward_min_interval_secs = config.node.discovery.forward_min_interval_secs;
let base_host_map = HostMap::from_peer_configs(config.peers());
let mut host_map = base_host_map.clone();
let hosts_path = std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH);
let hosts_file = HostMap::load_hosts_file(std::path::Path::new(
crate::upper::hosts::DEFAULT_HOSTS_PATH,
));
host_map.merge(hosts_file);
let host_map = Arc::new(host_map);
let host_map =
reloadable::HostMapReloadable::new(base_host_map.clone(), hosts_path.clone());
let peer_acl = acl::PeerAclReloader::with_alias_sources(
std::path::PathBuf::from(acl::DEFAULT_PEERS_ALLOW_PATH),
std::path::PathBuf::from(acl::DEFAULT_PEERS_DENY_PATH),
@@ -762,16 +762,14 @@ impl Node {
let coords_response_interval_ms = config.node.session.coords_response_interval_ms;
let base_host_map = HostMap::from_peer_configs(config.peers());
let mut host_map = base_host_map.clone();
host_map.merge(HostMap::load_hosts_file(std::path::Path::new(
crate::upper::hosts::DEFAULT_HOSTS_PATH,
)));
let host_map = Arc::new(host_map);
let hosts_path = std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH);
let host_map =
reloadable::HostMapReloadable::new(base_host_map.clone(), hosts_path.clone());
let peer_acl = acl::PeerAclReloader::with_alias_sources(
std::path::PathBuf::from(acl::DEFAULT_PEERS_ALLOW_PATH),
std::path::PathBuf::from(acl::DEFAULT_PEERS_DENY_PATH),
base_host_map,
std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH),
hosts_path,
);
#[cfg(unix)]
@@ -1094,6 +1092,13 @@ impl Node {
self.identity.npub()
}
/// Reload the host map if the backing `/etc/fips/hosts` file changed.
///
/// Returns `true` if a new snapshot was published.
pub(crate) async fn reload_host_map(&mut self) -> bool {
self.host_map.reload().await
}
/// Return a human-readable display name for a NodeAddr.
///
/// Lookup order:
@@ -1103,7 +1108,8 @@ impl Node {
/// 4. Session endpoint's short npub (end-to-end, may not be direct peer)
/// 5. Truncated NodeAddr hex (unknown address)
pub(crate) fn peer_display_name(&self, addr: &NodeAddr) -> String {
if let Some(hostname) = self.host_map.lookup_hostname(addr) {
let hosts = self.host_map.load();
if let Some(hostname) = hosts.lookup_hostname(addr) {
return hostname.to_string();
}
if let Some(name) = self.peer_aliases.get(addr) {
+360
View File
@@ -0,0 +1,360 @@
//! 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 dedup cache (`recent_requests`) is at capacity, so the
/// `LookupRequest` is dropped without being forwarded. Tracked via
/// [`DiscoveryStats::req_dedup_cache_full`](crate::node::stats::DiscoveryStats).
ReqDedupCacheFull,
/// 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::ReqDedupCacheFull,
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)
));
}
}
+322
View File
@@ -0,0 +1,322 @@
//! Lock-free reloadable configuration / resource snapshots.
//!
//! Several node-owned resources are loaded from disk at startup and may be
//! re-read when the backing file changes (for example the `/etc/fips/hosts`
//! map). Historically each one carried its own ad-hoc reloader with a
//! slightly different shape. The [`Reloadable`] trait normalizes them onto a
//! single contract built around an [`arc_swap::ArcSwap`] snapshot.
//!
//! # Canonical Arc-wrapper template
//!
//! These resources follow a single-writer / many-reader pattern: the node
//! tick task is the only writer, while the hot path reads the current value
//! frequently and must never block.
//!
//! - The reader-facing immutable snapshot lives in an
//! [`arc_swap::ArcSwap<T>`]. Readers call [`Reloadable::load`], which yields
//! a lock-free [`arc_swap::Guard<Arc<T>>`] that derefs straight to the
//! snapshot — no mutex, no clone on the read path.
//! - The owning struct also holds the change-detection state (file mtime,
//! immutable base data, source path). That state is touched only by
//! [`Reloadable::reload`], which runs on the single writer task.
//! - `reload` builds a brand-new `T` and then stores `Arc::new(new)` into the
//! `ArcSwap`, so a reader either sees the entire old snapshot or the entire
//! new one — never a partial update.
//! - Construction performs the initial synchronous load so the snapshot is
//! valid before the node starts serving reads.
//!
//! `reload` returns `bool` rather than a value or `Result`: the underlying
//! loaders already absorb I/O errors internally (a missing or unreadable
//! source degrades to an empty/base snapshot plus a warning), so callers have
//! nothing to handle. The return flag reports only whether the snapshot was
//! replaced, which is all the tick loop needs for logging.
//!
//! # What is and isn't `Reloadable`
//!
//! Two node-owned resources are deliberately *not* `Reloadable` because
//! neither has a "re-read from a backing source" concept:
//!
//! - `path_mtu_lookup` is an event-driven cache (`Arc<RwLock<HashMap>>`)
//! populated from observed path-MTU discovery traffic, not loaded from a
//! file. There is nothing to poll. (Its read side could adopt the same
//! lock-free `ArcSwap` shape in the future, but that is an optimization, not
//! a reload.)
//! - `nostr_discovery` is an async spawned subsystem, not a snapshot of disk
//! state.
//!
//! Both [`HostMapReloadable`] and the peer ACL reloader currently stat
//! `/etc/fips/hosts` independently each tick (the ACL reloader embeds its own
//! hosts reloader for alias resolution). A single small-file `stat` per tick
//! is cheap, so the duplicate is left in place; collapsing the two onto a
//! single shared mtime observation is a possible future cleanup.
use std::sync::Arc;
use crate::upper::hosts::{HostMap, file_mtime};
/// A resource backed by a lock-free [`arc_swap::ArcSwap`] snapshot that can be
/// re-read from its source on demand.
///
/// See the [module documentation](self) for the canonical Arc-wrapper
/// template that implementors follow.
pub trait Reloadable: Send {
/// The immutable snapshot type readers observe.
type Snapshot;
/// Re-read the backing source and replace the snapshot if it changed.
///
/// Returns `true` if a new snapshot was stored, `false` if nothing
/// changed. I/O errors are absorbed internally (degrading to an
/// empty/base snapshot with a warning) rather than surfaced.
///
/// Driven once per node tick from the rx loop, alongside the other
/// reloadable resources.
async fn reload(&mut self) -> bool;
/// Acquire a lock-free guard over the current snapshot.
///
/// This is the hot-path read: it performs no locking and no allocation.
fn load(&self) -> arc_swap::Guard<Arc<Self::Snapshot>>;
}
/// Reloadable hostname → npub map (base peer aliases merged with the operator
/// hosts file).
///
/// Holds the immutable base map (from peer-config aliases) plus the
/// change-detection state for the hosts file. The effective map (base merged
/// with the hosts file) is published through an [`arc_swap::ArcSwap`] so the
/// display path can read it without locking.
pub struct HostMapReloadable {
/// Reader-facing effective snapshot (base merged with hosts file).
snapshot: arc_swap::ArcSwap<HostMap>,
/// Base map from peer-config aliases (never changes). Read only by
/// `reload` on the tick task.
base: HostMap,
/// Path to the operator hosts file. Read only by `reload`.
path: std::path::PathBuf,
/// Last observed modification time of the hosts file (`None` if absent).
/// Read only by `reload`.
last_mtime: Option<std::time::SystemTime>,
}
impl HostMapReloadable {
/// Create a reloadable host map.
///
/// Performs the initial load of the hosts file and merges it over the
/// base map so the published snapshot is valid immediately.
pub fn new(base: HostMap, path: std::path::PathBuf) -> Self {
let last_mtime = file_mtime(&path);
let hosts_file = HostMap::load_hosts_file(&path);
let mut effective = base.clone();
effective.merge(hosts_file);
Self {
snapshot: arc_swap::ArcSwap::from(Arc::new(effective)),
base,
path,
last_mtime,
}
}
}
impl Reloadable for HostMapReloadable {
type Snapshot = HostMap;
async fn reload(&mut self) -> bool {
let current_mtime = file_mtime(&self.path);
if current_mtime == self.last_mtime {
return false;
}
// File appeared, disappeared, or was modified.
self.last_mtime = current_mtime;
let hosts_file = HostMap::load_hosts_file(&self.path);
let mut new_effective = self.base.clone();
new_effective.merge(hosts_file);
let count = new_effective.len();
self.snapshot.store(Arc::new(new_effective));
tracing::info!(
path = %self.path.display(),
entries = count,
"Reloaded hosts file"
);
true
}
fn load(&self) -> arc_swap::Guard<Arc<HostMap>> {
self.snapshot.load()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::Identity;
#[tokio::test]
async fn test_initial_load_base_and_file() {
let id_base = Identity::generate();
let id_file = Identity::generate();
let mut base = HostMap::new();
base.insert("core", &id_base.npub()).unwrap();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
std::fs::write(&path, format!("gateway {}\n", id_file.npub())).unwrap();
let reloadable = HostMapReloadable::new(base, path);
let snapshot = reloadable.load();
assert_eq!(snapshot.len(), 2);
assert!(snapshot.lookup_npub("core").is_some());
assert!(snapshot.lookup_npub("gateway").is_some());
}
#[tokio::test]
async fn test_initial_load_no_file_base_only() {
let id = Identity::generate();
let mut base = HostMap::new();
base.insert("core", &id.npub()).unwrap();
let reloadable =
HostMapReloadable::new(base, std::path::PathBuf::from("/nonexistent/hosts"));
let snapshot = reloadable.load();
assert_eq!(snapshot.len(), 1);
assert!(snapshot.lookup_npub("core").is_some());
}
#[tokio::test]
async fn test_reload_detects_file_change() {
let id1 = Identity::generate();
let id2 = Identity::generate();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
std::fs::write(&path, format!("gateway {}\n", id1.npub())).unwrap();
let mut reloadable = HostMapReloadable::new(HostMap::new(), path.clone());
assert_eq!(reloadable.load().len(), 1);
assert_eq!(
reloadable.load().lookup_npub("gateway"),
Some(id1.npub().as_str())
);
// No change yet.
assert!(!reloadable.reload().await);
// Bump mtime by rewriting; sleep for filesystem mtime granularity.
std::thread::sleep(std::time::Duration::from_millis(50));
std::fs::write(
&path,
format!("gateway {}\nnew-host {}\n", id1.npub(), id2.npub()),
)
.unwrap();
assert!(reloadable.reload().await);
let snapshot = reloadable.load();
assert_eq!(snapshot.len(), 2);
assert!(snapshot.lookup_npub("new-host").is_some());
}
#[tokio::test]
async fn test_reload_no_change_returns_false() {
let id = Identity::generate();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
std::fs::write(&path, format!("gateway {}\n", id.npub())).unwrap();
let mut reloadable = HostMapReloadable::new(HostMap::new(), path);
assert!(!reloadable.reload().await);
assert!(!reloadable.reload().await);
}
#[tokio::test]
async fn test_reload_detects_file_deletion() {
let id = Identity::generate();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
std::fs::write(&path, format!("gateway {}\n", id.npub())).unwrap();
let mut reloadable = HostMapReloadable::new(HostMap::new(), path.clone());
assert_eq!(reloadable.load().len(), 1);
std::fs::remove_file(&path).unwrap();
assert!(reloadable.reload().await);
assert!(reloadable.load().is_empty());
}
#[tokio::test]
async fn test_reload_detects_file_creation() {
let id = Identity::generate();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
let mut reloadable = HostMapReloadable::new(HostMap::new(), path.clone());
assert!(reloadable.load().is_empty());
std::fs::write(&path, format!("gateway {}\n", id.npub())).unwrap();
assert!(reloadable.reload().await);
let snapshot = reloadable.load();
assert_eq!(snapshot.len(), 1);
assert!(snapshot.lookup_npub("gateway").is_some());
}
#[tokio::test]
async fn test_reload_preserves_base() {
let id_base = Identity::generate();
let id_file = Identity::generate();
let mut base = HostMap::new();
base.insert("core", &id_base.npub()).unwrap();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
std::fs::write(&path, format!("gateway {}\n", id_file.npub())).unwrap();
let mut reloadable = HostMapReloadable::new(base, path.clone());
assert_eq!(reloadable.load().len(), 2);
std::fs::remove_file(&path).unwrap();
assert!(reloadable.reload().await);
let snapshot = reloadable.load();
assert_eq!(snapshot.len(), 1);
assert!(snapshot.lookup_npub("core").is_some());
assert!(snapshot.lookup_npub("gateway").is_none());
}
/// The initial published snapshot must be byte-for-byte equivalent to the
/// pre-migration `Arc<HostMap>` built by `base.clone()` + `merge(file)`.
#[tokio::test]
async fn test_initial_snapshot_matches_pre_migration_construction() {
let id_base = Identity::generate();
let id_file = Identity::generate();
let mut base = HostMap::new();
base.insert("core", &id_base.npub()).unwrap();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("hosts");
std::fs::write(&path, format!("gateway {}\n", id_file.npub())).unwrap();
// Pre-migration construction.
let mut expected = base.clone();
expected.merge(HostMap::load_hosts_file(&path));
// Post-migration construction.
let reloadable = HostMapReloadable::new(base, path);
let snapshot = reloadable.load();
assert_eq!(snapshot.len(), expected.len());
for key in ["core", "gateway"] {
assert_eq!(snapshot.lookup_npub(key), expected.lookup_npub(key));
}
}
}
+558
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,
@@ -107,6 +130,7 @@ pub struct DiscoveryStats {
pub req_received: u64,
pub req_decode_error: u64,
pub req_duplicate: u64,
pub req_dedup_cache_full: u64,
pub req_target_is_us: u64,
pub req_forwarded: u64,
pub req_ttl_exhausted: u64,
@@ -123,16 +147,31 @@ 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::ReqDedupCacheFull => self.req_dedup_cache_full += 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,
req_decode_error: self.req_decode_error,
req_duplicate: self.req_duplicate,
req_dedup_cache_full: self.req_dedup_cache_full,
req_target_is_us: self.req_target_is_us,
req_forwarded: self.req_forwarded,
req_ttl_exhausted: self.req_ttl_exhausted,
@@ -148,6 +187,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 +204,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 +213,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 +229,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 +237,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 +271,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 +299,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 +511,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 +530,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) ---
@@ -352,6 +588,7 @@ pub struct DiscoveryStatsSnapshot {
pub req_received: u64,
pub req_decode_error: u64,
pub req_duplicate: u64,
pub req_dedup_cache_full: u64,
pub req_target_is_us: u64,
pub req_forwarded: u64,
pub req_ttl_exhausted: u64,
@@ -367,6 +604,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 +617,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 +625,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 +646,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 +690,301 @@ 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_dedup_cache_full() {
let mut s = DiscoveryStats::default();
s.record_reject(DiscoveryReject::ReqDedupCacheFull);
assert_eq!(s.req_dedup_cache_full, 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);
}
}
+32 -4
View File
@@ -1,7 +1,9 @@
use super::*;
use crate::ReceivedPacket;
use crate::node::acl::PeerAclReloader;
use crate::node::reloadable::HostMapReloadable;
use crate::node::wire::{build_msg1, build_msg2};
use crate::upper::hosts::HostMap;
use crate::utils::index::SessionIndex;
use std::path::PathBuf;
use std::time::Duration;
@@ -29,7 +31,7 @@ async fn test_outbound_connect_denied_by_denylist() {
let (dir, mut node) = make_acl_node();
let denied = Identity::generate();
std::fs::write(deny_path(&dir), format!("{}\n", denied.npub())).unwrap();
node.reload_peer_acl();
node.reload_peer_acl().await;
let result = node
.initiate_connection(
@@ -51,7 +53,7 @@ async fn test_inbound_msg1_denied_by_acl() {
let node_a = make_node();
std::fs::write(deny_path(&dir), format!("{}\n", node_a.npub())).unwrap();
node_b.reload_peer_acl();
node_b.reload_peer_acl().await;
let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity.pubkey_full());
let mut conn_a = PeerConnection::outbound(LinkId::new(1), peer_b_identity, 1000);
@@ -121,7 +123,7 @@ async fn test_outbound_msg2_denied_after_acl_reload() {
let wire_msg2 = build_msg2(our_index_b, our_index_a, &noise_msg2);
std::fs::write(deny_path(&dir), format!("{}\n", node_b.npub())).unwrap();
assert!(node_a.reload_peer_acl());
assert!(node_a.reload_peer_acl().await);
let packet = ReceivedPacket::with_timestamp(transport_id, remote_addr, wire_msg2, 1100);
node_a.handle_msg2(packet).await;
@@ -132,13 +134,39 @@ async fn test_outbound_msg2_denied_after_acl_reload() {
assert!(node_a.pending_outbound.is_empty());
}
#[tokio::test]
async fn test_host_map_hot_reloads_from_tick() {
let dir = tempfile::tempdir().unwrap();
let hosts_path = dir.path().join("hosts");
let mut node = Node::new(Config::new()).unwrap();
node.host_map = HostMapReloadable::new(HostMap::new(), hosts_path.clone());
let peer = Identity::generate();
let peer_addr = *PeerIdentity::from_pubkey_full(peer.pubkey_full()).node_addr();
// No hosts file yet: the display name is not the alias.
assert_ne!(node.peer_display_name(&peer_addr), "gateway");
assert!(!node.reload_host_map().await);
// Write a hosts entry and let the tick-driven reload pick it up.
std::thread::sleep(Duration::from_millis(50));
std::fs::write(&hosts_path, format!("gateway {}\n", peer.npub())).unwrap();
assert!(node.reload_host_map().await);
assert_eq!(node.peer_display_name(&peer_addr), "gateway");
// No further change: reload reports nothing replaced.
assert!(!node.reload_host_map().await);
}
#[tokio::test]
async fn test_outbound_connect_not_denied_by_allowlist_miss() {
let (dir, mut node) = make_acl_node();
let denied = Identity::generate();
let allowed = Identity::generate();
std::fs::write(allow_path(&dir), format!("{}\n", allowed.npub())).unwrap();
node.reload_peer_acl();
node.reload_peer_acl().await;
let result = node
.initiate_connection(
+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);
}
+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(),