1use super::*;
2
3impl Node {
4 pub fn identity(&self) -> &Identity {
8 &self.identity
9 }
10
11 pub fn node_addr(&self) -> &NodeAddr {
13 self.identity.node_addr()
14 }
15
16 pub fn npub(&self) -> String {
18 self.identity.npub()
19 }
20
21 pub(crate) fn peer_display_name(&self, addr: &NodeAddr) -> String {
30 if let Some(hostname) = self.host_map.lookup_hostname(addr) {
31 return hostname.to_string();
32 }
33 if let Some(name) = self.peer_aliases.get(addr) {
34 return name.clone();
35 }
36 if let Some(peer) = self.peers.get(addr) {
37 return peer.identity().short_npub();
38 }
39 if let Some(entry) = self.sessions.get(addr) {
40 let (xonly, _) = entry.remote_pubkey().x_only_public_key();
41 return PeerIdentity::from_pubkey(xonly).short_npub();
42 }
43 addr.short_hex()
44 }
45
46 pub(in crate::node) fn deregister_session_index(&mut self, cache_key: (TransportId, u32)) {
48 let removed_index = self.peers.remove_session_index_with_owner_state(&cache_key);
52 let _ = removed_index;
53 }
54
55 pub(in crate::node) fn ensure_current_session_index_registered(
64 &mut self,
65 node_addr: &NodeAddr,
66 context: &'static str,
67 ) -> bool {
68 match self
69 .peers
70 .ensure_current_session_index_registered(node_addr)
71 {
72 CurrentSessionIndexRegistration::MissingActivePeer => false,
73 CurrentSessionIndexRegistration::MissingTransportId => {
74 warn!(
75 peer = %self.peer_display_name(node_addr),
76 context,
77 "Cannot register current session index without transport id"
78 );
79 false
80 }
81 CurrentSessionIndexRegistration::MissingLocalIndex => {
82 warn!(
83 peer = %self.peer_display_name(node_addr),
84 context,
85 "Cannot register current session index without local index"
86 );
87 false
88 }
89 CurrentSessionIndexRegistration::AlreadyRegistered(_) => true,
90 CurrentSessionIndexRegistration::Repaired(registered) => {
91 if let Some(existing) = registered.previous_owner {
92 warn!(
93 peer = %self.peer_display_name(node_addr),
94 previous_owner = %self.peer_display_name(&existing),
95 transport_id = %registered.session_index.key.0,
96 our_index = %registered.session_index.index,
97 context,
98 "Repairing current session index with stale owner"
99 );
100 } else {
101 warn!(
102 peer = %self.peer_display_name(node_addr),
103 transport_id = %registered.session_index.key.0,
104 our_index = %registered.session_index.index,
105 context,
106 "Repairing missing current session index"
107 );
108 }
109 true
110 }
111 }
112 }
113
114 pub(in crate::node) fn log_active_peer_insert_result(
115 &self,
116 node_addr: &NodeAddr,
117 inserted: &InsertedActivePeer,
118 context: &'static str,
119 ) {
120 if let Some(previous_peer) = inserted.previous_peer.as_ref() {
121 debug!(
122 peer = %self.peer_display_name(node_addr),
123 previous_link_id = %previous_peer.link_id(),
124 context,
125 "Replaced active peer storage during lifecycle insert"
126 );
127 }
128
129 match inserted.current_session_index {
130 Some(registered) => {
131 if let Some(previous_owner) = registered.previous_owner {
132 debug!(
133 peer = %self.peer_display_name(node_addr),
134 previous_owner = %self.peer_display_name(&previous_owner),
135 transport_id = %registered.session_index.key.0,
136 our_index = %registered.session_index.index,
137 context,
138 "Replaced current session-index owner during lifecycle insert"
139 );
140 }
141 }
142 None => {
143 warn!(
144 peer = %self.peer_display_name(node_addr),
145 context,
146 "Inserted active peer without a current session index"
147 );
148 }
149 }
150 }
151
152 pub(in crate::node) fn log_active_peer_session_replacement_result(
153 &self,
154 node_addr: &NodeAddr,
155 replacement: &ReplacedActivePeerCurrentSession,
156 context: &'static str,
157 ) {
158 if replacement.replay_suppressed_count > 0 {
159 debug!(
160 peer = %self.peer_display_name(node_addr),
161 count = replacement.replay_suppressed_count,
162 context,
163 "Suppressed replay detections during link transition"
164 );
165 }
166
167 if let Some(previous_owner) = replacement.new_session_index.previous_owner {
168 debug!(
169 peer = %self.peer_display_name(node_addr),
170 previous_owner = %self.peer_display_name(&previous_owner),
171 transport_id = %replacement.new_session_index.session_index.key.0,
172 our_index = %replacement.new_session_index.session_index.index,
173 context,
174 "Replaced current session-index owner during session replacement"
175 );
176 }
177 }
178
179 pub(in crate::node) fn log_registered_peer_session_index_result(
180 &self,
181 node_addr: &NodeAddr,
182 registered: &RegisteredPeerSessionIndex,
183 context: &'static str,
184 ) {
185 if let Some(previous_owner) = registered.previous_owner {
186 debug!(
187 peer = %self.peer_display_name(node_addr),
188 previous_owner = %self.peer_display_name(&previous_owner),
189 transport_id = %registered.session_index.key.0,
190 our_index = %registered.session_index.index,
191 index_kind = ?registered.session_index.kind,
192 context,
193 "Replaced session-index owner during lifecycle registration"
194 );
195 }
196 }
197
198 pub fn config(&self) -> &Config {
202 &self.config
203 }
204
205 pub fn effective_ipv6_mtu(&self) -> u16 {
211 crate::upper::icmp::effective_ipv6_mtu(self.transport_mtu())
212 }
213
214 pub fn transport_mtu(&self) -> u16 {
231 let min_operational = self
232 .transports
233 .values()
234 .filter(|h| h.is_operational())
235 .map(|h| h.mtu())
236 .min();
237 if let Some(mtu) = min_operational {
238 return mtu;
239 }
240 if let Some((_, cfg)) = self.config.transports.udp.iter().next() {
242 return cfg.mtu();
243 }
244 1280
245 }
246
247 pub fn state(&self) -> NodeState {
251 self.state
252 }
253
254 pub fn uptime(&self) -> std::time::Duration {
256 self.started_at.elapsed()
257 }
258
259 pub fn is_running(&self) -> bool {
261 self.state.is_operational()
262 }
263
264 pub fn is_leaf_only(&self) -> bool {
266 self.is_leaf_only
267 }
268
269 pub fn tree_state(&self) -> &TreeState {
273 &self.tree_state
274 }
275
276 pub fn tree_state_mut(&mut self) -> &mut TreeState {
278 &mut self.tree_state
279 }
280
281 pub fn bloom_state(&self) -> &BloomState {
285 &self.bloom_state
286 }
287
288 pub fn bloom_state_mut(&mut self) -> &mut BloomState {
290 &mut self.bloom_state
291 }
292
293 pub fn estimated_mesh_size(&self) -> Option<u64> {
297 self.estimated_mesh_size
298 }
299
300 pub(crate) fn compute_mesh_size(&mut self) {
307 let my_addr = *self.tree_state.my_node_addr();
308 let parent_id = *self.tree_state.my_declaration().parent_id();
309 let is_root = self.tree_state.is_root();
310
311 let max_fpr = self.config.node.bloom.max_inbound_fpr;
312 let mut child_count: u32 = 0;
313 let mut union: Option<BloomFilter> = None;
314
315 let add_to_union = |union: &mut Option<BloomFilter>, filter: &BloomFilter| match union {
316 None => *union = Some(filter.clone()),
317 Some(existing) => {
318 let _ = existing.merge(filter);
320 }
321 };
322
323 if !is_root
325 && let Some(parent) = self.peers.get(&parent_id)
326 && let Some(filter) = parent.inbound_filter()
327 {
328 add_to_union(&mut union, filter);
329 }
330
331 for (peer_addr, peer) in &self.peers {
334 if peer_addr == &parent_id {
335 continue;
336 }
337 if let Some(decl) = self.tree_state.peer_declaration(peer_addr)
338 && *decl.parent_id() == my_addr
339 {
340 child_count += 1;
341 if let Some(filter) = peer.inbound_filter() {
342 add_to_union(&mut union, filter);
343 }
344 }
345 }
346
347 let Some(mut union) = union else {
348 self.estimated_mesh_size = None;
349 return;
350 };
351 union.insert(&my_addr);
352
353 let Some(union_estimate) = union.estimated_count(max_fpr) else {
356 self.estimated_mesh_size = None;
357 return;
358 };
359
360 let size = union_estimate.round() as u64;
361 self.estimated_mesh_size = Some(size);
362
363 let now = std::time::Instant::now();
365 let should_log = match self.last_mesh_size_log {
366 None => true,
367 Some(last) => {
368 now.duration_since(last)
369 >= std::time::Duration::from_secs(self.config.node.mmp.log_interval_secs)
370 }
371 };
372 if should_log {
373 tracing::debug!(
374 estimated_mesh_size = size,
375 peers = self.peers.len(),
376 children = child_count,
377 "Mesh size estimate"
378 );
379 self.last_mesh_size_log = Some(now);
380 }
381 }
382
383 pub fn coord_cache(&self) -> &CoordCache {
387 &self.coord_cache
388 }
389
390 pub fn coord_cache_mut(&mut self) -> &mut CoordCache {
392 &mut self.coord_cache
393 }
394
395 pub fn stats(&self) -> &stats::NodeStats {
399 &self.stats
400 }
401
402 pub(crate) fn stats_mut(&mut self) -> &mut stats::NodeStats {
404 &mut self.stats
405 }
406
407 pub fn stats_history(&self) -> &stats_history::StatsHistory {
409 &self.stats_history
410 }
411
412 pub(crate) fn record_stats_history(&mut self) {
415 let fwd = &self.stats.forwarding;
416 let now = std::time::Instant::now();
417 let peers_with_mmp: Vec<f64> = self
418 .peers
419 .keys()
420 .filter_map(|addr| {
421 self.packet_mover2
422 .fmp_link_metrics(addr, now)
423 .map(|metrics| metrics.loss_rate)
424 })
425 .collect();
426 let loss_rate = if peers_with_mmp.is_empty() {
427 0.0
428 } else {
429 peers_with_mmp.iter().sum::<f64>() / peers_with_mmp.len() as f64
430 };
431
432 let snap = stats_history::Snapshot {
433 mesh_size: self.estimated_mesh_size,
434 tree_depth: self.tree_state.my_coords().depth() as u32,
435 peer_count: self.peers.len() as u64,
436 parent_switches_total: self.stats.tree.parent_switches,
437 bytes_in_total: fwd.received_bytes,
438 bytes_out_total: fwd.forwarded_bytes + fwd.originated_bytes,
439 packets_in_total: fwd.received_packets,
440 packets_out_total: fwd.forwarded_packets + fwd.originated_packets,
441 loss_rate,
442 active_sessions: self.sessions.len() as u64,
443 };
444
445 let peer_snaps: Vec<stats_history::PeerSnapshot> = self
446 .peers
447 .values()
448 .map(|p| {
449 let stats = p.link_stats();
450 let metrics = self.packet_mover2.fmp_link_metrics(p.node_addr(), now);
451 let srtt_ms = metrics.and_then(|metrics| metrics.srtt_ms);
452 let loss_rate = metrics.map(|metrics| metrics.loss_rate);
453 let ecn_ce = metrics.map_or(0, |metrics| metrics.ecn_ce_count as u64);
454 stats_history::PeerSnapshot {
455 node_addr: *p.node_addr(),
456 last_seen: now,
457 srtt_ms,
458 loss_rate,
459 bytes_in_total: stats.bytes_recv,
460 bytes_out_total: stats.bytes_sent,
461 packets_in_total: stats.packets_recv,
462 packets_out_total: stats.packets_sent,
463 ecn_ce_total: ecn_ce,
464 }
465 })
466 .collect();
467
468 self.stats_history.tick(now, &snap, &peer_snaps);
469 }
470
471 pub fn tun_state(&self) -> TunState {
475 self.tun_state
476 }
477
478 pub fn tun_name(&self) -> Option<&str> {
480 self.tun_name.as_deref()
481 }
482
483 pub fn set_max_connections(&mut self, max: usize) {
487 self.max_connections = max;
488 }
489
490 pub fn set_max_peers(&mut self, max: usize) {
492 self.max_peers = max;
493 }
494
495 pub(crate) fn outbound_admission_check(&self) -> bool {
498 let connection_used = self
499 .peers
500 .connection_len()
501 .saturating_add(self.pending_connects.len());
502 let peer_allowed = self.max_peers == 0 || self.peers.len() < self.max_peers;
503 let connection_allowed =
504 self.max_connections == 0 || connection_used < self.max_connections;
505 let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
506 peer_allowed && connection_allowed && link_allowed
507 }
508
509 pub(crate) fn open_discovery_outbound_admission_check(&self) -> bool {
513 if !self.outbound_admission_check() {
514 return false;
515 }
516
517 let nostr = &self.config.node.discovery.nostr;
518 if !nostr.enabled || nostr.policy != NostrDiscoveryPolicy::Open {
519 return true;
520 }
521
522 let configured_npubs = self
523 .config
524 .peers()
525 .iter()
526 .map(|peer| peer.npub.clone())
527 .collect::<HashSet<_>>();
528 self.open_discovery_enqueue_budget(&configured_npubs) > 0
529 }
530
531 pub(crate) fn outbound_direct_refresh_admission_check(&self) -> bool {
535 let connection_used = self
536 .peers
537 .connection_len()
538 .saturating_add(self.pending_connects.len());
539 let connection_allowed =
540 self.max_connections == 0 || connection_used < self.max_connections;
541 let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
542 connection_allowed && link_allowed
543 }
544
545 pub fn set_max_links(&mut self, max: usize) {
547 self.max_links = max;
548 }
549
550 pub fn connection_count(&self) -> usize {
554 self.peers.connection_len()
555 }
556
557 pub fn peer_count(&self) -> usize {
559 self.peers.len()
560 }
561
562 pub fn link_count(&self) -> usize {
564 self.links.len()
565 }
566
567 pub fn transport_count(&self) -> usize {
569 self.transports.len()
570 }
571
572 pub fn allocate_transport_id(&mut self) -> TransportId {
576 let id = TransportId::new(self.next_transport_id);
577 self.next_transport_id += 1;
578 id
579 }
580
581 pub fn get_transport(&self, id: &TransportId) -> Option<&TransportHandle> {
583 self.transports.get(id)
584 }
585
586 pub fn get_transport_mut(&mut self, id: &TransportId) -> Option<&mut TransportHandle> {
588 self.transports.get_mut(id)
589 }
590
591 pub fn transport_ids(&self) -> impl Iterator<Item = &TransportId> {
593 self.transports.keys()
594 }
595
596 pub fn packet_rx(&mut self) -> Option<&mut PacketRx> {
598 self.packet_rx.as_mut()
599 }
600
601 pub fn allocate_link_id(&mut self) -> LinkId {
605 let id = LinkId::new(self.next_link_id);
606 self.next_link_id += 1;
607 id
608 }
609
610 pub fn add_link(&mut self, link: Link) -> Result<(), NodeError> {
612 if self.max_links > 0 && self.links.len() >= self.max_links {
613 return Err(NodeError::MaxLinksExceeded {
614 max: self.max_links,
615 });
616 }
617 let link_id = link.link_id();
618
619 self.links.insert(link_id, link);
620 Ok(())
621 }
622
623 pub fn get_link(&self, link_id: &LinkId) -> Option<&Link> {
625 self.links.get(link_id)
626 }
627
628 pub fn get_link_mut(&mut self, link_id: &LinkId) -> Option<&mut Link> {
630 self.links.get_mut(link_id)
631 }
632
633 pub fn find_link_by_addr(
635 &self,
636 transport_id: TransportId,
637 addr: &TransportAddr,
638 ) -> Option<LinkId> {
639 self.links.lookup_addr(transport_id, addr)
640 }
641
642 pub fn remove_link(&mut self, link_id: &LinkId) -> Option<Link> {
648 self.links.remove(link_id)
649 }
650
651 pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
652 if !self.bootstrap_transports.contains(&transport_id) {
653 return;
654 }
655
656 let transport_in_use = self
657 .links
658 .values()
659 .any(|link| link.transport_id() == transport_id)
660 || self
661 .peers
662 .connection_values()
663 .any(|conn| conn.transport_id() == Some(transport_id))
664 || self
665 .peers
666 .values()
667 .any(|peer| peer.transport_id() == Some(transport_id))
668 || self
669 .pending_connects
670 .iter()
671 .any(|pending| pending.transport_id == transport_id);
672
673 if transport_in_use {
674 return;
675 }
676
677 tracing::debug!(
678 transport_id = %transport_id,
679 "bootstrap transport has no remaining references; dropping"
680 );
681
682 self.bootstrap_transports.remove(&transport_id);
683 self.transport_drops.remove(&transport_id);
684 self.transport_socket_drops.remove(&transport_id);
685 self.transport_namespace_drops.remove(&transport_id);
686 self.transports.remove(&transport_id);
687 self.udp_transport_resolution_cache.clear();
688 }
689
690 pub fn links(&self) -> impl Iterator<Item = &Link> {
692 self.links.values()
693 }
694
695 pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> {
699 let link_id = connection.link_id();
700
701 if self.peers.contains_connection(&link_id) {
702 return Err(NodeError::ConnectionAlreadyExists(link_id));
703 }
704
705 if self.max_connections > 0 && self.peers.connection_len() >= self.max_connections {
706 return Err(NodeError::MaxConnectionsExceeded {
707 max: self.max_connections,
708 });
709 }
710
711 self.peers.insert_connection(link_id, connection);
712 Ok(())
713 }
714
715 pub fn get_connection(&self, link_id: &LinkId) -> Option<&PeerConnection> {
717 self.peers.get_connection(link_id)
718 }
719
720 pub fn get_connection_mut(&mut self, link_id: &LinkId) -> Option<&mut PeerConnection> {
722 self.peers.get_connection_mut(link_id)
723 }
724
725 pub fn remove_connection(&mut self, link_id: &LinkId) -> Option<PeerConnection> {
727 self.peers.remove_connection(link_id)
728 }
729
730 pub fn connections(&self) -> impl Iterator<Item = &PeerConnection> {
732 self.peers.connection_values()
733 }
734
735 pub fn get_peer(&self, node_addr: &NodeAddr) -> Option<&ActivePeer> {
739 self.peers.get(node_addr)
740 }
741
742 pub fn get_peer_mut(&mut self, node_addr: &NodeAddr) -> Option<&mut ActivePeer> {
744 self.peers.get_mut(node_addr)
745 }
746
747 pub fn remove_peer(&mut self, node_addr: &NodeAddr) -> Option<ActivePeer> {
749 self.peers.remove(node_addr)
750 }
751
752 pub fn peers(&self) -> impl Iterator<Item = &ActivePeer> {
754 self.peers.values()
755 }
756
757 pub fn nostr_discovery_handle(&self) -> Option<&crate::discovery::nostr::NostrDiscovery> {
761 self.nostr_discovery.as_deref()
762 }
763
764 pub fn peer_ids(&self) -> impl Iterator<Item = &NodeAddr> {
766 self.peers.keys()
767 }
768
769 pub fn sendable_peers(&self) -> impl Iterator<Item = &ActivePeer> {
771 self.peers.values().filter(|p| p.can_send())
772 }
773
774 pub fn sendable_peer_count(&self) -> usize {
776 self.peers.values().filter(|p| p.can_send()).count()
777 }
778
779 pub(crate) fn set_discovery_fallback_transit_allowed(
780 &mut self,
781 peer_addr: NodeAddr,
782 allowed: bool,
783 ) {
784 self.discovery_fallback_transit
785 .set_allowed(peer_addr, allowed);
786 }
787
788 pub(crate) fn configured_discovery_fallback_transit(
789 &self,
790 peer_addr: &NodeAddr,
791 ) -> Option<bool> {
792 self.configured_peer(peer_addr)
793 .map(|peer| peer.discovery_fallback_transit)
794 }
795
796 pub(crate) fn configured_peer(&self, peer_addr: &NodeAddr) -> Option<&PeerConfig> {
797 self.configured_peer_send_weights.peer_config(peer_addr)
798 }
799
800 pub(in crate::node) fn active_peer_uses_configured_static_udp_path(
801 &self,
802 peer_addr: &NodeAddr,
803 ) -> bool {
804 let Some(peer_config) = self.configured_peer(peer_addr) else {
805 return false;
806 };
807
808 peer_config.addresses.iter().any(|candidate| {
809 candidate.seen_at_ms.is_none()
810 && candidate.transport.eq_ignore_ascii_case("udp")
811 && self.active_peer_matches_candidate(peer_addr, candidate)
812 })
813 }
814
815 pub(crate) fn discovery_fallback_transit_for_promotion(&self, peer_addr: &NodeAddr) -> bool {
816 if let Some(retry_state) = self.retry_pending.get(peer_addr) {
817 return retry_state.peer_config.discovery_fallback_transit;
818 }
819
820 if let Some(allowed) = self.configured_discovery_fallback_transit(peer_addr) {
821 return allowed;
822 }
823
824 self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open
825 }
826}