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)) {
58 let removed_index = self.peers.remove_session_index_with_owner_state(&cache_key);
62 let session_key = DecryptSessionKey::from(cache_key);
63 if self
64 .sessions
65 .unregister_worker_session_if_registered(&session_key)
66 && let Some(workers) = self.decrypt_workers.as_ref()
67 {
68 workers.unregister_session(session_key);
69 }
70 if let Some(removed_index) = removed_index
81 && !removed_index.owner_has_remaining_index
82 {
83 self.clear_connected_udp_for_peer(&removed_index.owner);
84 }
85 }
86
87 pub(in crate::node) fn ensure_current_session_index_registered(
96 &mut self,
97 node_addr: &NodeAddr,
98 context: &'static str,
99 ) -> bool {
100 match self
101 .peers
102 .ensure_current_session_index_registered(node_addr)
103 {
104 CurrentSessionIndexRegistration::MissingActivePeer => false,
105 CurrentSessionIndexRegistration::MissingTransportId => {
106 warn!(
107 peer = %self.peer_display_name(node_addr),
108 context,
109 "Cannot register current session index without transport id"
110 );
111 false
112 }
113 CurrentSessionIndexRegistration::MissingLocalIndex => {
114 warn!(
115 peer = %self.peer_display_name(node_addr),
116 context,
117 "Cannot register current session index without local index"
118 );
119 false
120 }
121 CurrentSessionIndexRegistration::AlreadyRegistered(_) => true,
122 CurrentSessionIndexRegistration::Repaired(registered) => {
123 if let Some(existing) = registered.previous_owner {
124 warn!(
125 peer = %self.peer_display_name(node_addr),
126 previous_owner = %self.peer_display_name(&existing),
127 transport_id = %registered.session_index.key.0,
128 our_index = %registered.session_index.index,
129 context,
130 "Repairing current session index with stale owner"
131 );
132 } else {
133 warn!(
134 peer = %self.peer_display_name(node_addr),
135 transport_id = %registered.session_index.key.0,
136 our_index = %registered.session_index.index,
137 context,
138 "Repairing missing current session index"
139 );
140 }
141 true
142 }
143 }
144 }
145
146 pub(in crate::node) fn log_active_peer_insert_result(
147 &self,
148 node_addr: &NodeAddr,
149 inserted: &InsertedActivePeer,
150 context: &'static str,
151 ) {
152 if let Some(previous_peer) = inserted.previous_peer.as_ref() {
153 debug!(
154 peer = %self.peer_display_name(node_addr),
155 previous_link_id = %previous_peer.link_id(),
156 context,
157 "Replaced active peer storage during lifecycle insert"
158 );
159 }
160
161 match inserted.current_session_index {
162 Some(registered) => {
163 if let Some(previous_owner) = registered.previous_owner {
164 debug!(
165 peer = %self.peer_display_name(node_addr),
166 previous_owner = %self.peer_display_name(&previous_owner),
167 transport_id = %registered.session_index.key.0,
168 our_index = %registered.session_index.index,
169 context,
170 "Replaced current session-index owner during lifecycle insert"
171 );
172 }
173 }
174 None => {
175 warn!(
176 peer = %self.peer_display_name(node_addr),
177 context,
178 "Inserted active peer without a current session index"
179 );
180 }
181 }
182 }
183
184 pub(in crate::node) fn log_active_peer_session_replacement_result(
185 &self,
186 node_addr: &NodeAddr,
187 replacement: &ReplacedActivePeerCurrentSession,
188 context: &'static str,
189 ) {
190 if replacement.replay_suppressed_count > 0 {
191 debug!(
192 peer = %self.peer_display_name(node_addr),
193 count = replacement.replay_suppressed_count,
194 context,
195 "Suppressed replay detections during link transition"
196 );
197 }
198
199 if let Some(previous_owner) = replacement.new_session_index.previous_owner {
200 debug!(
201 peer = %self.peer_display_name(node_addr),
202 previous_owner = %self.peer_display_name(&previous_owner),
203 transport_id = %replacement.new_session_index.session_index.key.0,
204 our_index = %replacement.new_session_index.session_index.index,
205 context,
206 "Replaced current session-index owner during session replacement"
207 );
208 }
209 }
210
211 pub(in crate::node) fn log_registered_peer_session_index_result(
212 &self,
213 node_addr: &NodeAddr,
214 registered: &RegisteredPeerSessionIndex,
215 context: &'static str,
216 ) {
217 if let Some(previous_owner) = registered.previous_owner {
218 debug!(
219 peer = %self.peer_display_name(node_addr),
220 previous_owner = %self.peer_display_name(&previous_owner),
221 transport_id = %registered.session_index.key.0,
222 our_index = %registered.session_index.index,
223 index_kind = ?registered.session_index.kind,
224 context,
225 "Replaced session-index owner during lifecycle registration"
226 );
227 }
228 }
229
230 pub fn config(&self) -> &Config {
234 &self.config
235 }
236
237 pub fn effective_ipv6_mtu(&self) -> u16 {
243 crate::upper::icmp::effective_ipv6_mtu(self.transport_mtu())
244 }
245
246 pub fn transport_mtu(&self) -> u16 {
263 let min_operational = self
264 .transports
265 .values()
266 .filter(|h| h.is_operational())
267 .map(|h| h.mtu())
268 .min();
269 if let Some(mtu) = min_operational {
270 return mtu;
271 }
272 if let Some((_, cfg)) = self.config.transports.udp.iter().next() {
274 return cfg.mtu();
275 }
276 1280
277 }
278
279 pub fn state(&self) -> NodeState {
283 self.state
284 }
285
286 pub fn uptime(&self) -> std::time::Duration {
288 self.started_at.elapsed()
289 }
290
291 pub fn is_running(&self) -> bool {
293 self.state.is_operational()
294 }
295
296 pub fn is_leaf_only(&self) -> bool {
298 self.is_leaf_only
299 }
300
301 pub fn tree_state(&self) -> &TreeState {
305 &self.tree_state
306 }
307
308 pub fn tree_state_mut(&mut self) -> &mut TreeState {
310 &mut self.tree_state
311 }
312
313 pub fn bloom_state(&self) -> &BloomState {
317 &self.bloom_state
318 }
319
320 pub fn bloom_state_mut(&mut self) -> &mut BloomState {
322 &mut self.bloom_state
323 }
324
325 pub fn estimated_mesh_size(&self) -> Option<u64> {
329 self.estimated_mesh_size
330 }
331
332 pub(crate) fn compute_mesh_size(&mut self) {
339 let my_addr = *self.tree_state.my_node_addr();
340 let parent_id = *self.tree_state.my_declaration().parent_id();
341 let is_root = self.tree_state.is_root();
342
343 let max_fpr = self.config.node.bloom.max_inbound_fpr;
344 let mut child_count: u32 = 0;
345 let mut union: Option<BloomFilter> = None;
346
347 let add_to_union = |union: &mut Option<BloomFilter>, filter: &BloomFilter| match union {
348 None => *union = Some(filter.clone()),
349 Some(existing) => {
350 let _ = existing.merge(filter);
352 }
353 };
354
355 if !is_root
357 && let Some(parent) = self.peers.get(&parent_id)
358 && let Some(filter) = parent.inbound_filter()
359 {
360 add_to_union(&mut union, filter);
361 }
362
363 for (peer_addr, peer) in &self.peers {
366 if peer_addr == &parent_id {
367 continue;
368 }
369 if let Some(decl) = self.tree_state.peer_declaration(peer_addr)
370 && *decl.parent_id() == my_addr
371 {
372 child_count += 1;
373 if let Some(filter) = peer.inbound_filter() {
374 add_to_union(&mut union, filter);
375 }
376 }
377 }
378
379 let Some(mut union) = union else {
380 self.estimated_mesh_size = None;
381 return;
382 };
383 union.insert(&my_addr);
384
385 let Some(union_estimate) = union.estimated_count(max_fpr) else {
388 self.estimated_mesh_size = None;
389 return;
390 };
391
392 let size = union_estimate.round() as u64;
393 self.estimated_mesh_size = Some(size);
394
395 let now = std::time::Instant::now();
397 let should_log = match self.last_mesh_size_log {
398 None => true,
399 Some(last) => {
400 now.duration_since(last)
401 >= std::time::Duration::from_secs(self.config.node.mmp.log_interval_secs)
402 }
403 };
404 if should_log {
405 tracing::debug!(
406 estimated_mesh_size = size,
407 peers = self.peers.len(),
408 children = child_count,
409 "Mesh size estimate"
410 );
411 self.last_mesh_size_log = Some(now);
412 }
413 }
414
415 pub fn coord_cache(&self) -> &CoordCache {
419 &self.coord_cache
420 }
421
422 pub fn coord_cache_mut(&mut self) -> &mut CoordCache {
424 &mut self.coord_cache
425 }
426
427 pub fn stats(&self) -> &stats::NodeStats {
431 &self.stats
432 }
433
434 pub(crate) fn stats_mut(&mut self) -> &mut stats::NodeStats {
436 &mut self.stats
437 }
438
439 pub fn stats_history(&self) -> &stats_history::StatsHistory {
441 &self.stats_history
442 }
443
444 pub(crate) fn record_stats_history(&mut self) {
447 let fwd = &self.stats.forwarding;
448 let peers_with_mmp: Vec<f64> = self
449 .peers
450 .values()
451 .filter_map(|p| p.mmp().map(|m| m.metrics.loss_rate()))
452 .collect();
453 let loss_rate = if peers_with_mmp.is_empty() {
454 0.0
455 } else {
456 peers_with_mmp.iter().sum::<f64>() / peers_with_mmp.len() as f64
457 };
458
459 let snap = stats_history::Snapshot {
460 mesh_size: self.estimated_mesh_size,
461 tree_depth: self.tree_state.my_coords().depth() as u32,
462 peer_count: self.peers.len() as u64,
463 parent_switches_total: self.stats.tree.parent_switches,
464 bytes_in_total: fwd.received_bytes,
465 bytes_out_total: fwd.forwarded_bytes + fwd.originated_bytes,
466 packets_in_total: fwd.received_packets,
467 packets_out_total: fwd.forwarded_packets + fwd.originated_packets,
468 loss_rate,
469 active_sessions: self.sessions.len() as u64,
470 };
471
472 let now = std::time::Instant::now();
473 let peer_snaps: Vec<stats_history::PeerSnapshot> = self
474 .peers
475 .values()
476 .map(|p| {
477 let stats = p.link_stats();
478 let (srtt_ms, loss_rate, ecn_ce) = match p.mmp() {
479 Some(m) => (
480 m.metrics.srtt_ms(),
481 Some(m.metrics.loss_rate()),
482 m.receiver.ecn_ce_count() as u64,
483 ),
484 None => (None, None, 0),
485 };
486 stats_history::PeerSnapshot {
487 node_addr: *p.node_addr(),
488 last_seen: now,
489 srtt_ms,
490 loss_rate,
491 bytes_in_total: stats.bytes_recv,
492 bytes_out_total: stats.bytes_sent,
493 packets_in_total: stats.packets_recv,
494 packets_out_total: stats.packets_sent,
495 ecn_ce_total: ecn_ce,
496 }
497 })
498 .collect();
499
500 self.stats_history.tick(now, &snap, &peer_snaps);
501 }
502
503 pub fn tun_state(&self) -> TunState {
507 self.tun_state
508 }
509
510 pub fn tun_name(&self) -> Option<&str> {
512 self.tun_name.as_deref()
513 }
514
515 pub fn set_max_connections(&mut self, max: usize) {
519 self.max_connections = max;
520 }
521
522 pub fn set_max_peers(&mut self, max: usize) {
524 self.max_peers = max;
525 }
526
527 pub(crate) fn outbound_admission_check(&self) -> bool {
530 let connection_used = self
531 .peers
532 .connection_len()
533 .saturating_add(self.pending_connects.len());
534 let peer_allowed = self.max_peers == 0 || self.peers.len() < self.max_peers;
535 let connection_allowed =
536 self.max_connections == 0 || connection_used < self.max_connections;
537 let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
538 peer_allowed && connection_allowed && link_allowed
539 }
540
541 pub(crate) fn open_discovery_outbound_admission_check(&self) -> bool {
545 if !self.outbound_admission_check() {
546 return false;
547 }
548
549 let nostr = &self.config.node.discovery.nostr;
550 if !nostr.enabled || nostr.policy != NostrDiscoveryPolicy::Open {
551 return true;
552 }
553
554 let configured_npubs = self
555 .config
556 .peers()
557 .iter()
558 .map(|peer| peer.npub.clone())
559 .collect::<HashSet<_>>();
560 self.open_discovery_enqueue_budget(&configured_npubs) > 0
561 }
562
563 pub(crate) fn outbound_direct_refresh_admission_check(&self) -> bool {
567 let connection_used = self
568 .peers
569 .connection_len()
570 .saturating_add(self.pending_connects.len());
571 let connection_allowed =
572 self.max_connections == 0 || connection_used < self.max_connections;
573 let link_allowed = self.max_links == 0 || self.links.len() < self.max_links;
574 connection_allowed && link_allowed
575 }
576
577 pub fn set_max_links(&mut self, max: usize) {
579 self.max_links = max;
580 }
581
582 pub fn connection_count(&self) -> usize {
586 self.peers.connection_len()
587 }
588
589 pub fn peer_count(&self) -> usize {
591 self.peers.len()
592 }
593
594 pub fn link_count(&self) -> usize {
596 self.links.len()
597 }
598
599 pub fn transport_count(&self) -> usize {
601 self.transports.len()
602 }
603
604 pub fn allocate_transport_id(&mut self) -> TransportId {
608 let id = TransportId::new(self.next_transport_id);
609 self.next_transport_id += 1;
610 id
611 }
612
613 pub fn get_transport(&self, id: &TransportId) -> Option<&TransportHandle> {
615 self.transports.get(id)
616 }
617
618 pub fn get_transport_mut(&mut self, id: &TransportId) -> Option<&mut TransportHandle> {
620 self.transports.get_mut(id)
621 }
622
623 pub fn transport_ids(&self) -> impl Iterator<Item = &TransportId> {
625 self.transports.keys()
626 }
627
628 pub fn packet_rx(&mut self) -> Option<&mut PacketRx> {
630 self.packet_rx.as_mut()
631 }
632
633 pub fn allocate_link_id(&mut self) -> LinkId {
637 let id = LinkId::new(self.next_link_id);
638 self.next_link_id += 1;
639 id
640 }
641
642 pub fn add_link(&mut self, link: Link) -> Result<(), NodeError> {
644 if self.max_links > 0 && self.links.len() >= self.max_links {
645 return Err(NodeError::MaxLinksExceeded {
646 max: self.max_links,
647 });
648 }
649 let link_id = link.link_id();
650
651 self.links.insert(link_id, link);
652 Ok(())
653 }
654
655 pub fn get_link(&self, link_id: &LinkId) -> Option<&Link> {
657 self.links.get(link_id)
658 }
659
660 pub fn get_link_mut(&mut self, link_id: &LinkId) -> Option<&mut Link> {
662 self.links.get_mut(link_id)
663 }
664
665 pub fn find_link_by_addr(
667 &self,
668 transport_id: TransportId,
669 addr: &TransportAddr,
670 ) -> Option<LinkId> {
671 self.links.lookup_addr(transport_id, addr)
672 }
673
674 pub fn remove_link(&mut self, link_id: &LinkId) -> Option<Link> {
680 self.links.remove(link_id)
681 }
682
683 pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
684 if !self.bootstrap_transports.contains(&transport_id) {
685 return;
686 }
687
688 let transport_in_use = self
689 .links
690 .values()
691 .any(|link| link.transport_id() == transport_id)
692 || self
693 .peers
694 .connection_values()
695 .any(|conn| conn.transport_id() == Some(transport_id))
696 || self
697 .peers
698 .values()
699 .any(|peer| peer.transport_id() == Some(transport_id))
700 || self
701 .pending_connects
702 .iter()
703 .any(|pending| pending.transport_id == transport_id);
704
705 if transport_in_use {
706 return;
707 }
708
709 tracing::debug!(
710 transport_id = %transport_id,
711 "bootstrap transport has no remaining references; dropping"
712 );
713
714 self.bootstrap_transports.remove(&transport_id);
715 self.transport_drops.remove(&transport_id);
716 self.transport_socket_drops.remove(&transport_id);
717 self.transport_namespace_drops.remove(&transport_id);
718 self.transports.remove(&transport_id);
719 }
720
721 pub fn links(&self) -> impl Iterator<Item = &Link> {
723 self.links.values()
724 }
725
726 pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> {
730 let link_id = connection.link_id();
731
732 if self.peers.contains_connection(&link_id) {
733 return Err(NodeError::ConnectionAlreadyExists(link_id));
734 }
735
736 if self.max_connections > 0 && self.peers.connection_len() >= self.max_connections {
737 return Err(NodeError::MaxConnectionsExceeded {
738 max: self.max_connections,
739 });
740 }
741
742 self.peers.insert_connection(link_id, connection);
743 Ok(())
744 }
745
746 pub fn get_connection(&self, link_id: &LinkId) -> Option<&PeerConnection> {
748 self.peers.get_connection(link_id)
749 }
750
751 pub fn get_connection_mut(&mut self, link_id: &LinkId) -> Option<&mut PeerConnection> {
753 self.peers.get_connection_mut(link_id)
754 }
755
756 pub fn remove_connection(&mut self, link_id: &LinkId) -> Option<PeerConnection> {
758 self.peers.remove_connection(link_id)
759 }
760
761 pub fn connections(&self) -> impl Iterator<Item = &PeerConnection> {
763 self.peers.connection_values()
764 }
765
766 pub fn get_peer(&self, node_addr: &NodeAddr) -> Option<&ActivePeer> {
770 self.peers.get(node_addr)
771 }
772
773 pub fn get_peer_mut(&mut self, node_addr: &NodeAddr) -> Option<&mut ActivePeer> {
775 self.peers.get_mut(node_addr)
776 }
777
778 pub fn remove_peer(&mut self, node_addr: &NodeAddr) -> Option<ActivePeer> {
780 self.peers.remove(node_addr)
781 }
782
783 pub fn peers(&self) -> impl Iterator<Item = &ActivePeer> {
785 self.peers.values()
786 }
787
788 pub fn nostr_discovery_handle(&self) -> Option<&crate::discovery::nostr::NostrDiscovery> {
792 self.nostr_discovery.as_deref()
793 }
794
795 pub fn peer_ids(&self) -> impl Iterator<Item = &NodeAddr> {
797 self.peers.keys()
798 }
799
800 pub fn sendable_peers(&self) -> impl Iterator<Item = &ActivePeer> {
802 self.peers.values().filter(|p| p.can_send())
803 }
804
805 pub fn sendable_peer_count(&self) -> usize {
807 self.peers.values().filter(|p| p.can_send()).count()
808 }
809
810 pub(crate) fn set_discovery_fallback_transit_allowed(
811 &mut self,
812 peer_addr: NodeAddr,
813 allowed: bool,
814 ) {
815 self.discovery_fallback_transit
816 .set_allowed(peer_addr, allowed);
817 }
818
819 pub(crate) fn configured_discovery_fallback_transit(
820 &self,
821 peer_addr: &NodeAddr,
822 ) -> Option<bool> {
823 self.configured_peer(peer_addr)
824 .map(|peer| peer.discovery_fallback_transit)
825 }
826
827 pub(crate) fn configured_peer(&self, peer_addr: &NodeAddr) -> Option<&PeerConfig> {
828 self.configured_peer_send_weights.peer_config(peer_addr)
829 }
830
831 pub(in crate::node) fn active_peer_uses_configured_static_udp_path(
832 &self,
833 peer_addr: &NodeAddr,
834 ) -> bool {
835 let Some(peer_config) = self.configured_peer(peer_addr) else {
836 return false;
837 };
838
839 peer_config.addresses.iter().any(|candidate| {
840 candidate.seen_at_ms.is_none()
841 && candidate.transport.eq_ignore_ascii_case("udp")
842 && self.active_peer_matches_candidate(peer_addr, candidate)
843 })
844 }
845
846 pub(crate) fn discovery_fallback_transit_for_promotion(&self, peer_addr: &NodeAddr) -> bool {
847 if let Some(retry_state) = self.retry_pending.get(peer_addr) {
848 return retry_state.peer_config.discovery_fallback_transit;
849 }
850
851 if let Some(allowed) = self.configured_discovery_fallback_transit(peer_addr) {
852 return allowed;
853 }
854
855 self.config.node.discovery.nostr.policy != crate::config::NostrDiscoveryPolicy::Open
856 }
857}