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(in crate::node) fn runtime_tun_mtu(&self) -> u16 {
221 self.config.tun.mtu.unwrap_or_else(|| {
222 self.effective_ipv6_mtu()
223 .max(crate::upper::config::DEFAULT_TUN_MTU)
224 })
225 }
226
227 pub fn transport_mtu(&self) -> u16 {
244 let min_operational = self
245 .transports
246 .values()
247 .filter(|h| h.is_operational())
248 .map(|h| h.mtu())
249 .min();
250 if let Some(mtu) = min_operational {
251 return mtu;
252 }
253 if let Some((_, cfg)) = self.config.transports.udp.iter().next() {
255 return cfg.mtu();
256 }
257 1280
258 }
259
260 pub fn state(&self) -> NodeState {
264 self.state
265 }
266
267 pub fn uptime(&self) -> std::time::Duration {
269 self.started_at.elapsed()
270 }
271
272 pub fn is_running(&self) -> bool {
274 self.state.is_operational()
275 }
276
277 pub fn is_leaf_only(&self) -> bool {
279 self.is_leaf_only
280 }
281
282 pub fn tree_state(&self) -> &TreeState {
286 &self.tree_state
287 }
288
289 pub fn tree_state_mut(&mut self) -> &mut TreeState {
291 &mut self.tree_state
292 }
293
294 pub fn bloom_state(&self) -> &BloomState {
298 &self.bloom_state
299 }
300
301 pub fn bloom_state_mut(&mut self) -> &mut BloomState {
303 &mut self.bloom_state
304 }
305
306 pub fn estimated_mesh_size(&self) -> Option<u64> {
310 self.estimated_mesh_size
311 }
312
313 pub(crate) fn compute_mesh_size(&mut self) {
320 let my_addr = *self.tree_state.my_node_addr();
321 let parent_id = *self.tree_state.my_declaration().parent_id();
322 let is_root = self.tree_state.is_root();
323
324 let max_fpr = self.config.node.bloom.max_inbound_fpr;
325 let mut child_count: u32 = 0;
326 let mut union: Option<BloomFilter> = None;
327
328 let add_to_union = |union: &mut Option<BloomFilter>, filter: &BloomFilter| match union {
329 None => *union = Some(filter.clone()),
330 Some(existing) => {
331 let _ = existing.merge(filter);
333 }
334 };
335
336 if !is_root
338 && let Some(parent) = self.peers.get(&parent_id)
339 && let Some(filter) = parent.inbound_filter()
340 {
341 add_to_union(&mut union, filter);
342 }
343
344 for (peer_addr, peer) in &self.peers {
347 if peer_addr == &parent_id {
348 continue;
349 }
350 if let Some(decl) = self.tree_state.peer_declaration(peer_addr)
351 && *decl.parent_id() == my_addr
352 {
353 child_count += 1;
354 if let Some(filter) = peer.inbound_filter() {
355 add_to_union(&mut union, filter);
356 }
357 }
358 }
359
360 let Some(mut union) = union else {
361 self.estimated_mesh_size = None;
362 return;
363 };
364 union.insert(&my_addr);
365
366 let Some(union_estimate) = union.estimated_count(max_fpr) else {
369 self.estimated_mesh_size = None;
370 return;
371 };
372
373 let size = union_estimate.round() as u64;
374 self.estimated_mesh_size = Some(size);
375
376 let now = std::time::Instant::now();
378 let should_log = match self.last_mesh_size_log {
379 None => true,
380 Some(last) => {
381 now.duration_since(last)
382 >= std::time::Duration::from_secs(self.config.node.mmp.log_interval_secs)
383 }
384 };
385 if should_log {
386 tracing::debug!(
387 estimated_mesh_size = size,
388 peers = self.peers.len(),
389 children = child_count,
390 "Mesh size estimate"
391 );
392 self.last_mesh_size_log = Some(now);
393 }
394 }
395
396 pub fn coord_cache(&self) -> &CoordCache {
400 &self.coord_cache
401 }
402
403 pub fn coord_cache_mut(&mut self) -> &mut CoordCache {
405 &mut self.coord_cache
406 }
407
408 pub fn stats(&self) -> &stats::NodeStats {
412 &self.stats
413 }
414
415 pub(crate) fn stats_mut(&mut self) -> &mut stats::NodeStats {
417 &mut self.stats
418 }
419
420 pub fn stats_history(&self) -> &stats_history::StatsHistory {
422 &self.stats_history
423 }
424
425 pub(crate) fn record_stats_history(&mut self) {
428 let fwd = &self.stats.forwarding;
429 let now = std::time::Instant::now();
430 let peers_with_mmp: Vec<f64> = self
431 .peers
432 .keys()
433 .filter_map(|addr| {
434 self.dataplane
435 .fmp_link_metrics(addr, now)
436 .map(|metrics| metrics.loss_rate)
437 })
438 .collect();
439 let loss_rate = if peers_with_mmp.is_empty() {
440 0.0
441 } else {
442 peers_with_mmp.iter().sum::<f64>() / peers_with_mmp.len() as f64
443 };
444
445 let snap = stats_history::Snapshot {
446 mesh_size: self.estimated_mesh_size,
447 tree_depth: self.tree_state.my_coords().depth() as u32,
448 peer_count: self.peers.len() as u64,
449 parent_switches_total: self.stats.tree.parent_switches,
450 bytes_in_total: fwd.received_bytes,
451 bytes_out_total: fwd.forwarded_bytes + fwd.originated_bytes,
452 packets_in_total: fwd.received_packets,
453 packets_out_total: fwd.forwarded_packets + fwd.originated_packets,
454 loss_rate,
455 active_sessions: self.sessions.len() as u64,
456 };
457
458 let peer_snaps: Vec<stats_history::PeerSnapshot> = self
459 .peers
460 .values()
461 .map(|p| {
462 let stats = p.link_stats();
463 let metrics = self.dataplane.fmp_link_metrics(p.node_addr(), now);
464 let srtt_ms = metrics.and_then(|metrics| metrics.srtt_ms);
465 let loss_rate = metrics.map(|metrics| metrics.loss_rate);
466 let ecn_ce = metrics.map_or(0, |metrics| metrics.ecn_ce_count as u64);
467 stats_history::PeerSnapshot {
468 node_addr: *p.node_addr(),
469 last_seen: now,
470 srtt_ms,
471 loss_rate,
472 bytes_in_total: stats.bytes_recv,
473 bytes_out_total: stats.bytes_sent,
474 packets_in_total: stats.packets_recv,
475 packets_out_total: stats.packets_sent,
476 ecn_ce_total: ecn_ce,
477 }
478 })
479 .collect();
480
481 self.stats_history.tick(now, &snap, &peer_snaps);
482 }
483
484 pub fn tun_state(&self) -> TunState {
488 self.tun_state
489 }
490
491 pub fn tun_name(&self) -> Option<&str> {
493 self.tun_name.as_deref()
494 }
495
496 pub fn set_max_connections(&mut self, max: usize) {
500 self.max_connections = max;
501 }
502
503 pub fn set_max_peers(&mut self, max: usize) {
505 self.max_peers = max;
506 }
507
508 pub(crate) fn outbound_admission_check(&self) -> bool {
511 let connection_used = self
512 .peers
513 .connection_len()
514 .saturating_add(self.pending_connects.len());
515 let peer_allowed = self.max_peers == 0 || self.peers.len() < self.max_peers;
516 let connection_allowed =
517 self.max_connections == 0 || connection_used < self.max_connections;
518 let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
519 peer_allowed && connection_allowed && link_allowed
520 }
521
522 pub(crate) fn open_discovery_outbound_admission_check(&self) -> bool {
526 if !self.outbound_admission_check() {
527 return false;
528 }
529
530 let nostr = &self.config.node.discovery.nostr;
531 if !nostr.enabled || nostr.policy != NostrDiscoveryPolicy::Open {
532 return true;
533 }
534
535 let configured_npubs = self
536 .config
537 .peers()
538 .iter()
539 .map(|peer| peer.npub.clone())
540 .collect::<HashSet<_>>();
541 self.open_discovery_enqueue_budget(&configured_npubs) > 0
542 }
543
544 pub(crate) fn outbound_direct_refresh_admission_check(&self) -> bool {
548 let connection_used = self
549 .peers
550 .connection_len()
551 .saturating_add(self.pending_connects.len());
552 let connection_allowed =
553 self.max_connections == 0 || connection_used < self.max_connections;
554 let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
555 connection_allowed && link_allowed
556 }
557
558 pub fn set_max_links(&mut self, max: usize) {
560 self.max_links = max;
561 }
562
563 pub fn connection_count(&self) -> usize {
567 self.peers.connection_len()
568 }
569
570 pub fn peer_count(&self) -> usize {
572 self.peers.len()
573 }
574
575 pub fn link_count(&self) -> usize {
577 self.links.len()
578 }
579
580 pub fn transport_count(&self) -> usize {
582 self.transports.len()
583 }
584
585 pub fn allocate_transport_id(&mut self) -> TransportId {
589 let id = TransportId::new(self.next_transport_id);
590 self.next_transport_id += 1;
591 id
592 }
593
594 pub fn get_transport(&self, id: &TransportId) -> Option<&TransportHandle> {
596 self.transports.get(id)
597 }
598
599 pub fn get_transport_mut(&mut self, id: &TransportId) -> Option<&mut TransportHandle> {
601 self.transports.get_mut(id)
602 }
603
604 pub fn transport_ids(&self) -> impl Iterator<Item = &TransportId> {
606 self.transports.keys()
607 }
608
609 pub fn packet_rx(&mut self) -> Option<&mut PacketRx> {
611 self.packet_rx.as_mut()
612 }
613
614 pub fn allocate_link_id(&mut self) -> LinkId {
618 let id = LinkId::new(self.next_link_id);
619 self.next_link_id += 1;
620 id
621 }
622
623 pub fn add_link(&mut self, link: Link) -> Result<(), NodeError> {
625 if self.max_links > 0 && self.links.len() >= self.max_links {
626 return Err(NodeError::MaxLinksExceeded {
627 max: self.max_links,
628 });
629 }
630 let link_id = link.link_id();
631
632 self.links.insert(link_id, link);
633 Ok(())
634 }
635
636 pub fn get_link(&self, link_id: &LinkId) -> Option<&Link> {
638 self.links.get(link_id)
639 }
640
641 pub fn get_link_mut(&mut self, link_id: &LinkId) -> Option<&mut Link> {
643 self.links.get_mut(link_id)
644 }
645
646 pub fn find_link_by_addr(
648 &self,
649 transport_id: TransportId,
650 addr: &TransportAddr,
651 ) -> Option<LinkId> {
652 self.links.lookup_addr(transport_id, addr)
653 }
654
655 pub fn remove_link(&mut self, link_id: &LinkId) -> Option<Link> {
661 self.links.remove(link_id)
662 }
663
664 pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
665 if !self.bootstrap_transports.contains(&transport_id) {
666 return;
667 }
668
669 let transport_in_use = self
670 .links
671 .values()
672 .any(|link| link.transport_id() == transport_id)
673 || self
674 .peers
675 .connection_values()
676 .any(|conn| conn.transport_id() == Some(transport_id))
677 || self
678 .peers
679 .values()
680 .any(|peer| peer.transport_id() == Some(transport_id))
681 || self
682 .pending_connects
683 .iter()
684 .any(|pending| pending.transport_id == transport_id);
685
686 if transport_in_use {
687 return;
688 }
689
690 tracing::debug!(
691 transport_id = %transport_id,
692 "bootstrap transport has no remaining references; dropping"
693 );
694
695 self.bootstrap_transports.remove(&transport_id);
696 self.transport_drops.remove(&transport_id);
697 self.transport_socket_drops.remove(&transport_id);
698 self.transport_namespace_drops.remove(&transport_id);
699 self.transports.remove(&transport_id);
700 self.udp_transport_resolution_cache.clear();
701 }
702
703 pub fn links(&self) -> impl Iterator<Item = &Link> {
705 self.links.values()
706 }
707
708 pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> {
712 let link_id = connection.link_id();
713
714 if self.peers.contains_connection(&link_id) {
715 return Err(NodeError::ConnectionAlreadyExists(link_id));
716 }
717
718 if self.max_connections > 0 && self.peers.connection_len() >= self.max_connections {
719 return Err(NodeError::MaxConnectionsExceeded {
720 max: self.max_connections,
721 });
722 }
723
724 self.peers.insert_connection(link_id, connection);
725 Ok(())
726 }
727
728 pub fn get_connection(&self, link_id: &LinkId) -> Option<&PeerConnection> {
730 self.peers.get_connection(link_id)
731 }
732
733 pub fn get_connection_mut(&mut self, link_id: &LinkId) -> Option<&mut PeerConnection> {
735 self.peers.get_connection_mut(link_id)
736 }
737
738 pub fn remove_connection(&mut self, link_id: &LinkId) -> Option<PeerConnection> {
740 self.peers.remove_connection(link_id)
741 }
742
743 pub fn connections(&self) -> impl Iterator<Item = &PeerConnection> {
745 self.peers.connection_values()
746 }
747
748 pub fn get_peer(&self, node_addr: &NodeAddr) -> Option<&ActivePeer> {
752 self.peers.get(node_addr)
753 }
754
755 pub fn get_peer_mut(&mut self, node_addr: &NodeAddr) -> Option<&mut ActivePeer> {
757 self.peers.get_mut(node_addr)
758 }
759
760 pub fn remove_peer(&mut self, node_addr: &NodeAddr) -> Option<ActivePeer> {
762 self.peers.remove(node_addr)
763 }
764
765 pub fn peers(&self) -> impl Iterator<Item = &ActivePeer> {
767 self.peers.values()
768 }
769
770 pub fn nostr_discovery_handle(&self) -> Option<&crate::discovery::nostr::NostrDiscovery> {
774 self.nostr_discovery.as_deref()
775 }
776
777 pub fn peer_ids(&self) -> impl Iterator<Item = &NodeAddr> {
779 self.peers.keys()
780 }
781
782 pub fn sendable_peers(&self) -> impl Iterator<Item = &ActivePeer> {
784 self.peers.values().filter(|p| p.can_send())
785 }
786
787 pub fn sendable_peer_count(&self) -> usize {
789 self.peers.values().filter(|p| p.can_send()).count()
790 }
791
792 pub(crate) fn set_discovery_fallback_transit_allowed(
793 &mut self,
794 peer_addr: NodeAddr,
795 allowed: bool,
796 ) {
797 self.discovery_fallback_transit
798 .set_allowed(peer_addr, allowed);
799 }
800
801 pub(crate) fn configured_discovery_fallback_transit(
802 &self,
803 peer_addr: &NodeAddr,
804 ) -> Option<bool> {
805 self.configured_peer(peer_addr)
806 .map(|peer| peer.discovery_fallback_transit)
807 }
808
809 pub(crate) fn configured_peer(&self, peer_addr: &NodeAddr) -> Option<&PeerConfig> {
810 self.configured_peer_send_weights.peer_config(peer_addr)
811 }
812
813 pub(in crate::node) fn active_peer_uses_configured_static_udp_path(
814 &self,
815 peer_addr: &NodeAddr,
816 ) -> bool {
817 let Some(peer_config) = self.configured_peer(peer_addr) else {
818 return false;
819 };
820
821 peer_config.addresses.iter().any(|candidate| {
822 candidate.seen_at_ms.is_none()
823 && candidate.transport.eq_ignore_ascii_case("udp")
824 && self.active_peer_matches_candidate(peer_addr, candidate)
825 })
826 }
827
828 pub(crate) fn discovery_fallback_transit_for_promotion(&self, peer_addr: &NodeAddr) -> bool {
829 if let Some(retry_state) = self.retry_pending.get(peer_addr) {
830 return retry_state.peer_config.discovery_fallback_transit;
831 }
832
833 if let Some(allowed) = self.configured_discovery_fallback_transit(peer_addr) {
834 return allowed;
835 }
836
837 self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open
838 }
839}