use super::*;
impl Node {
pub fn identity(&self) -> &Identity {
&self.identity
}
pub fn node_addr(&self) -> &NodeAddr {
self.identity.node_addr()
}
pub fn npub(&self) -> String {
self.identity.npub()
}
pub(crate) fn peer_display_name(&self, addr: &NodeAddr) -> String {
if let Some(hostname) = self.host_map.lookup_hostname(addr) {
return hostname.to_string();
}
if let Some(name) = self.peer_aliases.get(addr) {
return name.clone();
}
if let Some(peer) = self.peers.get(addr) {
return peer.identity().short_npub();
}
if let Some(entry) = self.sessions.get(addr) {
let (xonly, _) = entry.remote_pubkey().x_only_public_key();
return PeerIdentity::from_pubkey(xonly).short_npub();
}
addr.short_hex()
}
pub(in crate::node) fn deregister_session_index(&mut self, cache_key: (TransportId, u32)) {
let removed_index = self.peers.remove_session_index_with_owner_state(&cache_key);
let _ = removed_index;
}
pub(in crate::node) fn ensure_current_session_index_registered(
&mut self,
node_addr: &NodeAddr,
context: &'static str,
) -> bool {
match self
.peers
.ensure_current_session_index_registered(node_addr)
{
CurrentSessionIndexRegistration::MissingActivePeer => false,
CurrentSessionIndexRegistration::MissingTransportId => {
warn!(
peer = %self.peer_display_name(node_addr),
context,
"Cannot register current session index without transport id"
);
false
}
CurrentSessionIndexRegistration::MissingLocalIndex => {
warn!(
peer = %self.peer_display_name(node_addr),
context,
"Cannot register current session index without local index"
);
false
}
CurrentSessionIndexRegistration::AlreadyRegistered(_) => true,
CurrentSessionIndexRegistration::Repaired(registered) => {
if let Some(existing) = registered.previous_owner {
warn!(
peer = %self.peer_display_name(node_addr),
previous_owner = %self.peer_display_name(&existing),
transport_id = %registered.session_index.key.0,
our_index = %registered.session_index.index,
context,
"Repairing current session index with stale owner"
);
} else {
warn!(
peer = %self.peer_display_name(node_addr),
transport_id = %registered.session_index.key.0,
our_index = %registered.session_index.index,
context,
"Repairing missing current session index"
);
}
true
}
}
}
pub(in crate::node) fn log_active_peer_insert_result(
&self,
node_addr: &NodeAddr,
inserted: &InsertedActivePeer,
context: &'static str,
) {
if let Some(previous_peer) = inserted.previous_peer.as_ref() {
debug!(
peer = %self.peer_display_name(node_addr),
previous_link_id = %previous_peer.link_id(),
context,
"Replaced active peer storage during lifecycle insert"
);
}
match inserted.current_session_index {
Some(registered) => {
if let Some(previous_owner) = registered.previous_owner {
debug!(
peer = %self.peer_display_name(node_addr),
previous_owner = %self.peer_display_name(&previous_owner),
transport_id = %registered.session_index.key.0,
our_index = %registered.session_index.index,
context,
"Replaced current session-index owner during lifecycle insert"
);
}
}
None => {
warn!(
peer = %self.peer_display_name(node_addr),
context,
"Inserted active peer without a current session index"
);
}
}
}
pub(in crate::node) fn log_active_peer_session_replacement_result(
&self,
node_addr: &NodeAddr,
replacement: &ReplacedActivePeerCurrentSession,
context: &'static str,
) {
if replacement.replay_suppressed_count > 0 {
debug!(
peer = %self.peer_display_name(node_addr),
count = replacement.replay_suppressed_count,
context,
"Suppressed replay detections during link transition"
);
}
if let Some(previous_owner) = replacement.new_session_index.previous_owner {
debug!(
peer = %self.peer_display_name(node_addr),
previous_owner = %self.peer_display_name(&previous_owner),
transport_id = %replacement.new_session_index.session_index.key.0,
our_index = %replacement.new_session_index.session_index.index,
context,
"Replaced current session-index owner during session replacement"
);
}
}
pub(in crate::node) fn log_registered_peer_session_index_result(
&self,
node_addr: &NodeAddr,
registered: &RegisteredPeerSessionIndex,
context: &'static str,
) {
if let Some(previous_owner) = registered.previous_owner {
debug!(
peer = %self.peer_display_name(node_addr),
previous_owner = %self.peer_display_name(&previous_owner),
transport_id = %registered.session_index.key.0,
our_index = %registered.session_index.index,
index_kind = ?registered.session_index.kind,
context,
"Replaced session-index owner during lifecycle registration"
);
}
}
pub fn config(&self) -> &Config {
&self.config
}
pub fn effective_ipv6_mtu(&self) -> u16 {
crate::upper::icmp::effective_ipv6_mtu(self.transport_mtu())
}
pub(in crate::node) fn runtime_tun_mtu(&self) -> u16 {
self.config.tun.mtu.unwrap_or_else(|| {
self.effective_ipv6_mtu()
.max(crate::upper::config::DEFAULT_TUN_MTU)
})
}
pub fn transport_mtu(&self) -> u16 {
let min_operational = self
.transports
.values()
.filter(|h| h.is_operational())
.map(|h| h.mtu())
.min();
if let Some(mtu) = min_operational {
return mtu;
}
if let Some((_, cfg)) = self.config.transports.udp.iter().next() {
return cfg.mtu();
}
1280
}
pub fn state(&self) -> NodeState {
self.state
}
pub fn uptime(&self) -> std::time::Duration {
self.started_at.elapsed()
}
pub fn is_running(&self) -> bool {
self.state.is_operational()
}
pub fn is_leaf_only(&self) -> bool {
self.is_leaf_only
}
pub fn tree_state(&self) -> &TreeState {
&self.tree_state
}
pub fn tree_state_mut(&mut self) -> &mut TreeState {
&mut self.tree_state
}
pub fn bloom_state(&self) -> &BloomState {
&self.bloom_state
}
pub fn bloom_state_mut(&mut self) -> &mut BloomState {
&mut self.bloom_state
}
pub fn estimated_mesh_size(&self) -> Option<u64> {
self.estimated_mesh_size
}
pub(crate) fn compute_mesh_size(&mut self) {
let my_addr = *self.tree_state.my_node_addr();
let parent_id = *self.tree_state.my_declaration().parent_id();
let is_root = self.tree_state.is_root();
let max_fpr = self.config.node.bloom.max_inbound_fpr;
let mut child_count: u32 = 0;
let mut union: Option<BloomFilter> = None;
let add_to_union = |union: &mut Option<BloomFilter>, filter: &BloomFilter| match union {
None => *union = Some(filter.clone()),
Some(existing) => {
let _ = existing.merge(filter);
}
};
if !is_root
&& let Some(parent) = self.peers.get(&parent_id)
&& let Some(filter) = parent.inbound_filter()
{
add_to_union(&mut union, filter);
}
for (peer_addr, peer) in &self.peers {
if peer_addr == &parent_id {
continue;
}
if let Some(decl) = self.tree_state.peer_declaration(peer_addr)
&& *decl.parent_id() == my_addr
{
child_count += 1;
if let Some(filter) = peer.inbound_filter() {
add_to_union(&mut union, filter);
}
}
}
let Some(mut union) = union else {
self.estimated_mesh_size = None;
return;
};
union.insert(&my_addr);
let Some(union_estimate) = union.estimated_count(max_fpr) else {
self.estimated_mesh_size = None;
return;
};
let size = union_estimate.round() as u64;
self.estimated_mesh_size = Some(size);
let now = std::time::Instant::now();
let should_log = match self.last_mesh_size_log {
None => true,
Some(last) => {
now.duration_since(last)
>= std::time::Duration::from_secs(self.config.node.mmp.log_interval_secs)
}
};
if should_log {
tracing::debug!(
estimated_mesh_size = size,
peers = self.peers.len(),
children = child_count,
"Mesh size estimate"
);
self.last_mesh_size_log = Some(now);
}
}
pub fn coord_cache(&self) -> &CoordCache {
&self.coord_cache
}
pub fn coord_cache_mut(&mut self) -> &mut CoordCache {
&mut self.coord_cache
}
pub fn stats(&self) -> &stats::NodeStats {
&self.stats
}
pub(crate) fn stats_mut(&mut self) -> &mut stats::NodeStats {
&mut self.stats
}
pub fn stats_history(&self) -> &stats_history::StatsHistory {
&self.stats_history
}
pub(crate) fn record_stats_history(&mut self) {
let fwd = &self.stats.forwarding;
let now = std::time::Instant::now();
let peers_with_mmp: Vec<f64> = self
.peers
.keys()
.filter_map(|addr| {
self.dataplane
.fmp_link_metrics(addr, now)
.map(|metrics| metrics.loss_rate)
})
.collect();
let loss_rate = if peers_with_mmp.is_empty() {
0.0
} else {
peers_with_mmp.iter().sum::<f64>() / peers_with_mmp.len() as f64
};
let snap = stats_history::Snapshot {
mesh_size: self.estimated_mesh_size,
tree_depth: self.tree_state.my_coords().depth() as u32,
peer_count: self.peers.len() as u64,
parent_switches_total: self.stats.tree.parent_switches,
bytes_in_total: fwd.received_bytes,
bytes_out_total: fwd.forwarded_bytes + fwd.originated_bytes,
packets_in_total: fwd.received_packets,
packets_out_total: fwd.forwarded_packets + fwd.originated_packets,
loss_rate,
active_sessions: self.sessions.len() as u64,
};
let peer_snaps: Vec<stats_history::PeerSnapshot> = self
.peers
.values()
.map(|p| {
let stats = p.link_stats();
let metrics = self.dataplane.fmp_link_metrics(p.node_addr(), now);
let srtt_ms = metrics.and_then(|metrics| metrics.srtt_ms);
let loss_rate = metrics.map(|metrics| metrics.loss_rate);
let ecn_ce = metrics.map_or(0, |metrics| metrics.ecn_ce_count as u64);
stats_history::PeerSnapshot {
node_addr: *p.node_addr(),
last_seen: now,
srtt_ms,
loss_rate,
bytes_in_total: stats.bytes_recv,
bytes_out_total: stats.bytes_sent,
packets_in_total: stats.packets_recv,
packets_out_total: stats.packets_sent,
ecn_ce_total: ecn_ce,
}
})
.collect();
self.stats_history.tick(now, &snap, &peer_snaps);
}
pub fn tun_state(&self) -> TunState {
self.tun_state
}
pub fn tun_name(&self) -> Option<&str> {
self.tun_name.as_deref()
}
pub fn set_max_connections(&mut self, max: usize) {
self.max_connections = max;
}
pub fn set_max_peers(&mut self, max: usize) {
self.max_peers = max;
}
pub(crate) fn outbound_admission_check(&self) -> bool {
let connection_used = self
.peers
.connection_len()
.saturating_add(self.pending_connects.len());
let peer_allowed = self.max_peers == 0 || self.peers.len() < self.max_peers;
let connection_allowed =
self.max_connections == 0 || connection_used < self.max_connections;
let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
peer_allowed && connection_allowed && link_allowed
}
pub(crate) fn open_discovery_outbound_admission_check(&self) -> bool {
if !self.outbound_admission_check() {
return false;
}
let nostr = &self.config.node.discovery.nostr;
if !nostr.enabled || nostr.policy != NostrDiscoveryPolicy::Open {
return true;
}
let configured_npubs = self
.config
.peers()
.iter()
.map(|peer| peer.npub.clone())
.collect::<HashSet<_>>();
self.open_discovery_enqueue_budget(&configured_npubs) > 0
}
pub(crate) fn outbound_direct_refresh_admission_check(&self) -> bool {
let connection_used = self
.peers
.connection_len()
.saturating_add(self.pending_connects.len());
let connection_allowed =
self.max_connections == 0 || connection_used < self.max_connections;
let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
connection_allowed && link_allowed
}
pub fn set_max_links(&mut self, max: usize) {
self.max_links = max;
}
pub fn connection_count(&self) -> usize {
self.peers.connection_len()
}
pub fn peer_count(&self) -> usize {
self.peers.len()
}
pub fn link_count(&self) -> usize {
self.links.len()
}
pub fn transport_count(&self) -> usize {
self.transports.len()
}
pub fn allocate_transport_id(&mut self) -> TransportId {
let id = TransportId::new(self.next_transport_id);
self.next_transport_id += 1;
id
}
pub fn get_transport(&self, id: &TransportId) -> Option<&TransportHandle> {
self.transports.get(id)
}
pub fn get_transport_mut(&mut self, id: &TransportId) -> Option<&mut TransportHandle> {
self.transports.get_mut(id)
}
pub fn transport_ids(&self) -> impl Iterator<Item = &TransportId> {
self.transports.keys()
}
pub fn packet_rx(&mut self) -> Option<&mut PacketRx> {
self.packet_rx.as_mut()
}
pub fn allocate_link_id(&mut self) -> LinkId {
let id = LinkId::new(self.next_link_id);
self.next_link_id += 1;
id
}
pub fn add_link(&mut self, link: Link) -> Result<(), NodeError> {
if self.max_links > 0 && self.links.len() >= self.max_links {
return Err(NodeError::MaxLinksExceeded {
max: self.max_links,
});
}
let link_id = link.link_id();
self.links.insert(link_id, link);
Ok(())
}
pub fn get_link(&self, link_id: &LinkId) -> Option<&Link> {
self.links.get(link_id)
}
pub fn get_link_mut(&mut self, link_id: &LinkId) -> Option<&mut Link> {
self.links.get_mut(link_id)
}
pub fn find_link_by_addr(
&self,
transport_id: TransportId,
addr: &TransportAddr,
) -> Option<LinkId> {
self.links.lookup_addr(transport_id, addr)
}
pub fn remove_link(&mut self, link_id: &LinkId) -> Option<Link> {
self.links.remove(link_id)
}
pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
if !self.bootstrap_transports.contains(&transport_id) {
return;
}
let transport_in_use = self
.links
.values()
.any(|link| link.transport_id() == transport_id)
|| self
.peers
.connection_values()
.any(|conn| conn.transport_id() == Some(transport_id))
|| self
.peers
.values()
.any(|peer| peer.transport_id() == Some(transport_id))
|| self
.pending_connects
.iter()
.any(|pending| pending.transport_id == transport_id);
if transport_in_use {
return;
}
tracing::debug!(
transport_id = %transport_id,
"bootstrap transport has no remaining references; dropping"
);
self.bootstrap_transports.remove(&transport_id);
self.transport_drops.remove(&transport_id);
self.transport_socket_drops.remove(&transport_id);
self.transport_namespace_drops.remove(&transport_id);
self.transports.remove(&transport_id);
self.udp_transport_resolution_cache.clear();
}
pub fn links(&self) -> impl Iterator<Item = &Link> {
self.links.values()
}
pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> {
let link_id = connection.link_id();
if self.peers.contains_connection(&link_id) {
return Err(NodeError::ConnectionAlreadyExists(link_id));
}
if self.max_connections > 0 && self.peers.connection_len() >= self.max_connections {
return Err(NodeError::MaxConnectionsExceeded {
max: self.max_connections,
});
}
self.peers.insert_connection(link_id, connection);
Ok(())
}
pub fn get_connection(&self, link_id: &LinkId) -> Option<&PeerConnection> {
self.peers.get_connection(link_id)
}
pub fn get_connection_mut(&mut self, link_id: &LinkId) -> Option<&mut PeerConnection> {
self.peers.get_connection_mut(link_id)
}
pub fn remove_connection(&mut self, link_id: &LinkId) -> Option<PeerConnection> {
self.peers.remove_connection(link_id)
}
pub fn connections(&self) -> impl Iterator<Item = &PeerConnection> {
self.peers.connection_values()
}
pub fn get_peer(&self, node_addr: &NodeAddr) -> Option<&ActivePeer> {
self.peers.get(node_addr)
}
pub fn get_peer_mut(&mut self, node_addr: &NodeAddr) -> Option<&mut ActivePeer> {
self.peers.get_mut(node_addr)
}
pub fn remove_peer(&mut self, node_addr: &NodeAddr) -> Option<ActivePeer> {
self.peers.remove(node_addr)
}
pub fn peers(&self) -> impl Iterator<Item = &ActivePeer> {
self.peers.values()
}
pub fn nostr_discovery_handle(&self) -> Option<&crate::discovery::nostr::NostrDiscovery> {
self.nostr_discovery.as_deref()
}
pub fn peer_ids(&self) -> impl Iterator<Item = &NodeAddr> {
self.peers.keys()
}
pub fn sendable_peers(&self) -> impl Iterator<Item = &ActivePeer> {
self.peers.values().filter(|p| p.can_send())
}
pub fn sendable_peer_count(&self) -> usize {
self.peers.values().filter(|p| p.can_send()).count()
}
pub(crate) fn set_discovery_fallback_transit_allowed(
&mut self,
peer_addr: NodeAddr,
allowed: bool,
) {
self.discovery_fallback_transit
.set_allowed(peer_addr, allowed);
}
pub(crate) fn configured_discovery_fallback_transit(
&self,
peer_addr: &NodeAddr,
) -> Option<bool> {
self.configured_peer(peer_addr)
.map(|peer| peer.discovery_fallback_transit)
}
pub(crate) fn configured_peer(&self, peer_addr: &NodeAddr) -> Option<&PeerConfig> {
self.configured_peer_send_weights.peer_config(peer_addr)
}
pub(in crate::node) fn active_peer_uses_configured_static_udp_path(
&self,
peer_addr: &NodeAddr,
) -> bool {
let Some(peer_config) = self.configured_peer(peer_addr) else {
return false;
};
peer_config.addresses.iter().any(|candidate| {
candidate.seen_at_ms.is_none()
&& candidate.transport.eq_ignore_ascii_case("udp")
&& self.active_peer_matches_candidate(peer_addr, candidate)
})
}
pub(crate) fn discovery_fallback_transit_for_promotion(&self, peer_addr: &NodeAddr) -> bool {
if let Some(retry_state) = self.retry_pending.get(peer_addr) {
return retry_state.peer_config.discovery_fallback_transit;
}
if let Some(allowed) = self.configured_discovery_fallback_transit(peer_addr) {
return allowed;
}
self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open
}
}