1use super::*;
2
3impl Node {
4 pub fn new(config: Config) -> Result<Self, NodeError> {
6 config.validate()?;
7 let identity = config.create_identity()?;
8 let node_addr = *identity.node_addr();
9 let is_leaf_only = config.is_leaf_only();
10
11 let (decrypt_fallback_tx, decrypt_fallback_rx) =
12 decrypt_worker::decrypt_worker_fallback_channels();
13 let decrypt_fallback_rx = Some(decrypt_fallback_rx);
14
15 let mut startup_epoch = [0u8; 8];
16 rand::rng().fill_bytes(&mut startup_epoch);
17
18 let mut bloom_state = if is_leaf_only {
19 BloomState::leaf_only(node_addr)
20 } else {
21 BloomState::new(node_addr)
22 };
23 bloom_state.set_update_debounce_ms(config.node.bloom.update_debounce_ms);
24
25 let tun_state = if config.tun.enabled {
26 TunState::Configured
27 } else {
28 TunState::Disabled
29 };
30
31 let mut tree_state = TreeState::new(node_addr);
33 tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
34 tree_state.set_hold_down(config.node.tree.hold_down_secs);
35 tree_state.set_flap_dampening(
36 config.node.tree.flap_threshold,
37 config.node.tree.flap_window_secs,
38 config.node.tree.flap_dampening_secs,
39 );
40 tree_state
41 .sign_declaration(&identity)
42 .expect("signing own declaration should never fail");
43
44 let coord_cache = CoordCache::new(
45 config.node.cache.coord_size,
46 config.node.cache.coord_ttl_secs * 1000,
47 );
48 let rl = &config.node.rate_limit;
49 let msg1_rate_limiter = HandshakeRateLimiter::with_params(
50 rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
51 config.node.limits.max_pending_inbound,
52 );
53
54 let max_connections = config.node.limits.max_connections;
55 let max_peers = config.node.limits.max_peers;
56 let max_links = config.node.limits.max_links;
57 let coords_response_interval_ms = config.node.session.coords_response_interval_ms;
58 let backoff_base_secs = config.node.discovery.backoff_base_secs;
59 let backoff_max_secs = config.node.discovery.backoff_max_secs;
60 let forward_min_interval_secs = config.node.discovery.forward_min_interval_secs;
61
62 let (host_map, peer_acl) = Self::host_map_and_peer_acl(&config);
63 let configured_peer_send_weights = ConfiguredPeerSendWeights::from_config(&config);
64
65 Ok(Self {
66 identity,
67 startup_epoch,
68 started_at: std::time::Instant::now(),
69 config,
70 state: NodeState::Created,
71 is_leaf_only,
72 tree_state,
73 bloom_state,
74 coord_cache,
75 learned_routes: LearnedRouteTable::default(),
76 session_direct_degradation: SessionDirectDegradation::default(),
77 recent_requests: RecentDiscoveryRequests::default(),
78 transports: HashMap::new(),
79 transport_drops: TransportDropTracker::default(),
80 transport_socket_drops: TransportDropTracker::default(),
81 transport_namespace_drops: TransportDropTracker::default(),
82 links: LinkRegistry::default(),
83 packet_tx: None,
84 packet_rx: None,
85 peers: PeerLifecycleRegistry::default(),
86 sessions: SessionRegistry::default(),
87 identity_cache: IdentityCache::default(),
88 pending_session_traffic: PendingSessionTrafficQueues::default(),
89 pending_lookups: handlers::discovery::PendingDiscoveryLookups::default(),
90 max_connections,
91 max_peers,
92 max_links,
93 next_link_id: 1,
94 next_transport_id: 1,
95 stats: stats::NodeStats::new(),
96 stats_history: stats_history::StatsHistory::new(),
97 tun_state,
98 tun_name: None,
99 tun_tx: None,
100 tun_outbound_rx: None,
101 external_packet_tx: None,
102 endpoint_priority_command_rx: None,
103 endpoint_command_rx: None,
104 endpoint_events: EndpointEventRuntime::default(),
105 endpoint_bulk_feedback_rx: None,
106 #[cfg(unix)]
107 endpoint_bulk_send_runtime: None,
108 encrypt_workers: None,
109 decrypt_workers: None,
110 decrypt_fallback_tx,
111 decrypt_fallback_rx,
112 tun_reader_handle: None,
113 tun_writer_handle: None,
114 #[cfg(target_os = "macos")]
115 tun_shutdown_fd: None,
116 dns_identity_rx: None,
117 dns_task: None,
118 index_allocator: IndexAllocator::new(),
119 pending_outbound: PendingOutboundHandshakes::default(),
120 msg1_rate_limiter,
121 icmp_rate_limiter: IcmpRateLimiter::new(),
122 routing_error_rate_limiter: RoutingErrorRateLimiter::new(),
123 coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval(
124 std::time::Duration::from_millis(coords_response_interval_ms),
125 ),
126 discovery_backoff: DiscoveryBackoff::with_params(backoff_base_secs, backoff_max_secs),
127 discovery_forward_limiter: DiscoveryForwardRateLimiter::with_interval(
128 std::time::Duration::from_secs(forward_min_interval_secs),
129 ),
130 pending_connects: Vec::new(),
131 retry_pending: retry::PendingRouteRetries::default(),
132 nostr_discovery: None,
133 nostr_discovery_started_at_ms: None,
134 lan_discovery: None,
135 local_instance_registry: None,
136 local_instance_started_at_ms: None,
137 last_local_instance_publish_ms: None,
138 last_local_instance_scan_ms: None,
139 startup_open_discovery_sweep_done: false,
140 bootstrap_transports: BootstrapTransports::default(),
141 discovery_fallback_transit: DiscoveryFallbackTransit::default(),
142 last_parent_reeval: None,
143 last_congestion_log: None,
144 estimated_mesh_size: None,
145 last_mesh_size_log: None,
146 last_self_warn: None,
147 local_send_failures: LocalSendFailures::default(),
148 last_rx_loop_maintenance_timeout_at: None,
149 peer_aliases: HashMap::new(),
150 configured_peer_send_weights,
151 peer_acl,
152 host_map,
153 path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())),
154 })
155 }
156
157 pub fn with_identity(identity: Identity, config: Config) -> Result<Self, NodeError> {
162 config.validate()?;
163 let node_addr = *identity.node_addr();
164
165 let (decrypt_fallback_tx, decrypt_fallback_rx) =
166 decrypt_worker::decrypt_worker_fallback_channels();
167 let decrypt_fallback_rx = Some(decrypt_fallback_rx);
168
169 let mut startup_epoch = [0u8; 8];
170 rand::rng().fill_bytes(&mut startup_epoch);
171
172 let tun_state = if config.tun.enabled {
173 TunState::Configured
174 } else {
175 TunState::Disabled
176 };
177
178 let mut tree_state = TreeState::new(node_addr);
180 tree_state.set_parent_hysteresis(config.node.tree.parent_hysteresis);
181 tree_state.set_hold_down(config.node.tree.hold_down_secs);
182 tree_state.set_flap_dampening(
183 config.node.tree.flap_threshold,
184 config.node.tree.flap_window_secs,
185 config.node.tree.flap_dampening_secs,
186 );
187 tree_state
188 .sign_declaration(&identity)
189 .expect("signing own declaration should never fail");
190
191 let mut bloom_state = BloomState::new(node_addr);
192 bloom_state.set_update_debounce_ms(config.node.bloom.update_debounce_ms);
193
194 let coord_cache = CoordCache::new(
195 config.node.cache.coord_size,
196 config.node.cache.coord_ttl_secs * 1000,
197 );
198 let rl = &config.node.rate_limit;
199 let msg1_rate_limiter = HandshakeRateLimiter::with_params(
200 rate_limit::TokenBucket::with_params(rl.handshake_burst, rl.handshake_rate),
201 config.node.limits.max_pending_inbound,
202 );
203
204 let max_connections = config.node.limits.max_connections;
205 let max_peers = config.node.limits.max_peers;
206 let max_links = config.node.limits.max_links;
207 let coords_response_interval_ms = config.node.session.coords_response_interval_ms;
208
209 let (host_map, peer_acl) = Self::host_map_and_peer_acl(&config);
210 let configured_peer_send_weights = ConfiguredPeerSendWeights::from_config(&config);
211
212 Ok(Self {
213 identity,
214 startup_epoch,
215 started_at: std::time::Instant::now(),
216 config,
217 state: NodeState::Created,
218 is_leaf_only: false,
219 tree_state,
220 bloom_state,
221 coord_cache,
222 learned_routes: LearnedRouteTable::default(),
223 session_direct_degradation: SessionDirectDegradation::default(),
224 recent_requests: RecentDiscoveryRequests::default(),
225 transports: HashMap::new(),
226 transport_drops: TransportDropTracker::default(),
227 transport_socket_drops: TransportDropTracker::default(),
228 transport_namespace_drops: TransportDropTracker::default(),
229 links: LinkRegistry::default(),
230 packet_tx: None,
231 packet_rx: None,
232 peers: PeerLifecycleRegistry::default(),
233 sessions: SessionRegistry::default(),
234 identity_cache: IdentityCache::default(),
235 pending_session_traffic: PendingSessionTrafficQueues::default(),
236 pending_lookups: handlers::discovery::PendingDiscoveryLookups::default(),
237 max_connections,
238 max_peers,
239 max_links,
240 next_link_id: 1,
241 next_transport_id: 1,
242 stats: stats::NodeStats::new(),
243 stats_history: stats_history::StatsHistory::new(),
244 tun_state,
245 tun_name: None,
246 tun_tx: None,
247 tun_outbound_rx: None,
248 external_packet_tx: None,
249 endpoint_priority_command_rx: None,
250 endpoint_command_rx: None,
251 endpoint_events: EndpointEventRuntime::default(),
252 endpoint_bulk_feedback_rx: None,
253 #[cfg(unix)]
254 endpoint_bulk_send_runtime: None,
255 encrypt_workers: None,
256 decrypt_workers: None,
257 decrypt_fallback_tx,
258 decrypt_fallback_rx,
259 tun_reader_handle: None,
260 tun_writer_handle: None,
261 #[cfg(target_os = "macos")]
262 tun_shutdown_fd: None,
263 dns_identity_rx: None,
264 dns_task: None,
265 index_allocator: IndexAllocator::new(),
266 pending_outbound: PendingOutboundHandshakes::default(),
267 msg1_rate_limiter,
268 icmp_rate_limiter: IcmpRateLimiter::new(),
269 routing_error_rate_limiter: RoutingErrorRateLimiter::new(),
270 coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval(
271 std::time::Duration::from_millis(coords_response_interval_ms),
272 ),
273 discovery_backoff: DiscoveryBackoff::new(),
274 discovery_forward_limiter: DiscoveryForwardRateLimiter::new(),
275 pending_connects: Vec::new(),
276 retry_pending: retry::PendingRouteRetries::default(),
277 nostr_discovery: None,
278 nostr_discovery_started_at_ms: None,
279 lan_discovery: None,
280 local_instance_registry: None,
281 local_instance_started_at_ms: None,
282 last_local_instance_publish_ms: None,
283 last_local_instance_scan_ms: None,
284 startup_open_discovery_sweep_done: false,
285 bootstrap_transports: BootstrapTransports::default(),
286 discovery_fallback_transit: DiscoveryFallbackTransit::default(),
287 last_parent_reeval: None,
288 last_congestion_log: None,
289 estimated_mesh_size: None,
290 last_mesh_size_log: None,
291 last_self_warn: None,
292 local_send_failures: LocalSendFailures::default(),
293 last_rx_loop_maintenance_timeout_at: None,
294 peer_aliases: HashMap::new(),
295 configured_peer_send_weights,
296 peer_acl,
297 host_map,
298 path_mtu_lookup: Arc::new(std::sync::RwLock::new(HashMap::new())),
299 })
300 }
301
302 pub fn leaf_only(config: Config) -> Result<Self, NodeError> {
304 let mut node = Self::new(config)?;
305 node.is_leaf_only = true;
306 node.bloom_state = BloomState::leaf_only(*node.identity.node_addr());
307 Ok(node)
308 }
309
310 pub(super) fn host_map_and_peer_acl(config: &Config) -> (Arc<HostMap>, acl::PeerAclReloader) {
311 let base_host_map = HostMap::from_peer_configs(config.peers());
312 if !config.node.system_files_enabled {
313 return (
314 Arc::new(base_host_map.clone()),
315 acl::PeerAclReloader::memory_only(base_host_map),
316 );
317 }
318
319 let mut host_map = base_host_map.clone();
320 let hosts_path = std::path::PathBuf::from(crate::upper::hosts::DEFAULT_HOSTS_PATH);
321 let hosts_file = HostMap::load_hosts_file(std::path::Path::new(
322 crate::upper::hosts::DEFAULT_HOSTS_PATH,
323 ));
324 host_map.merge(hosts_file);
325 let peer_acl = acl::PeerAclReloader::with_alias_sources(
326 std::path::PathBuf::from(acl::DEFAULT_PEERS_ALLOW_PATH),
327 std::path::PathBuf::from(acl::DEFAULT_PEERS_DENY_PATH),
328 base_host_map,
329 hosts_path,
330 );
331 (Arc::new(host_map), peer_acl)
332 }
333
334 #[cfg(unix)]
335 pub(super) fn send_weight_for_peer(&self, peer_addr: &NodeAddr) -> u8 {
336 self.configured_peer_send_weights.weight_for(peer_addr)
337 }
338
339 #[cfg(unix)]
340 pub(in crate::node) fn resolve_peer_runtime_route_decision(
341 &mut self,
342 dest_addr: &NodeAddr,
343 now_ms: u64,
344 ) -> Result<PeerRuntimeRouteDecision, PeerRuntimeRouteDecisionError> {
345 let Some(next_hop_addr) = self.find_next_hop(dest_addr).map(|peer| *peer.node_addr())
346 else {
347 return Err(PeerRuntimeRouteDecisionError::NoRoute {
348 dest_addr: *dest_addr,
349 });
350 };
351
352 let peer_snapshot = self
353 .peers
354 .prepare_peer_runtime_route_snapshot(&next_hop_addr)
355 .map_err(|error| PeerRuntimeRouteDecisionError::FmpPreparation {
356 next_hop_addr,
357 error,
358 })?;
359 let scheduling_weight = self.send_weight_for_peer(&next_hop_addr);
360 let direct_path_blocks_direct_payload = next_hop_addr == *dest_addr
361 && self.session_direct_path_blocks_direct_payload(dest_addr, now_ms);
362
363 Ok(PeerRuntimeRouteDecision::new(
364 next_hop_addr,
365 peer_snapshot,
366 scheduling_weight,
367 direct_path_blocks_direct_payload,
368 ))
369 }
370
371 pub(super) async fn create_transports(&mut self, packet_tx: &PacketTx) -> Vec<TransportHandle> {
375 let mut transports = Vec::new();
376
377 let udp_instances: Vec<_> = self
379 .config
380 .transports
381 .udp
382 .iter()
383 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
384 .collect();
385
386 for (name, udp_config) in udp_instances {
388 let transport_id = self.allocate_transport_id();
389 let udp = UdpTransport::new(transport_id, name, udp_config, packet_tx.clone());
390 transports.push(TransportHandle::Udp(udp));
391 }
392
393 #[cfg(feature = "sim-transport")]
394 {
395 let sim_instances: Vec<_> = self
396 .config
397 .transports
398 .sim
399 .iter()
400 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
401 .collect();
402
403 for (name, sim_config) in sim_instances {
404 let transport_id = self.allocate_transport_id();
405 let sim = crate::transport::sim::SimTransport::new(
406 transport_id,
407 name,
408 sim_config,
409 packet_tx.clone(),
410 );
411 transports.push(TransportHandle::Sim(sim));
412 }
413 }
414
415 #[cfg(any(target_os = "linux", target_os = "macos"))]
417 {
418 let eth_instances: Vec<_> = self
419 .config
420 .transports
421 .ethernet
422 .iter()
423 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
424 .collect();
425 let xonly = self.identity.pubkey();
426 for (name, eth_config) in eth_instances {
427 let mut eth_config = eth_config;
428 if eth_config.discovery_scope.is_none() {
429 eth_config.discovery_scope = self.lan_discovery_scope();
430 }
431 let transport_id = self.allocate_transport_id();
432 let mut eth =
433 EthernetTransport::new(transport_id, name, eth_config, packet_tx.clone());
434 eth.set_local_pubkey(xonly);
435 transports.push(TransportHandle::Ethernet(eth));
436 }
437 }
438
439 let tcp_instances: Vec<_> = self
441 .config
442 .transports
443 .tcp
444 .iter()
445 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
446 .collect();
447
448 for (name, tcp_config) in tcp_instances {
449 let transport_id = self.allocate_transport_id();
450 let tcp = TcpTransport::new(transport_id, name, tcp_config, packet_tx.clone());
451 transports.push(TransportHandle::Tcp(tcp));
452 }
453
454 let tor_instances: Vec<_> = self
456 .config
457 .transports
458 .tor
459 .iter()
460 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
461 .collect();
462
463 for (name, tor_config) in tor_instances {
464 let transport_id = self.allocate_transport_id();
465 let tor = TorTransport::new(transport_id, name, tor_config, packet_tx.clone());
466 transports.push(TransportHandle::Tor(tor));
467 }
468
469 let webrtc_instances: Vec<_> = self
470 .config
471 .transports
472 .webrtc
473 .iter()
474 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
475 .collect();
476
477 #[cfg(feature = "webrtc-transport")]
478 {
479 for (name, webrtc_config) in webrtc_instances {
480 let transport_id = self.allocate_transport_id();
481 match WebRtcTransport::new(
482 transport_id,
483 name,
484 webrtc_config,
485 packet_tx.clone(),
486 &self.identity,
487 &self.config.node.discovery.nostr,
488 ) {
489 Ok(webrtc) => transports.push(TransportHandle::WebRtc(Box::new(webrtc))),
490 Err(err) => {
491 warn!(
492 transport_id = %transport_id,
493 error = %err,
494 "failed to initialize WebRTC transport"
495 );
496 }
497 }
498 }
499 }
500 #[cfg(not(feature = "webrtc-transport"))]
501 if !webrtc_instances.is_empty() {
502 warn!("WebRTC transport configured but this build lacks WebRTC transport support");
503 }
504
505 #[cfg(bluer_available)]
507 {
508 let ble_instances: Vec<_> = self
509 .config
510 .transports
511 .ble
512 .iter()
513 .map(|(name, config)| (name.map(|s| s.to_string()), config.clone()))
514 .collect();
515
516 #[cfg(all(bluer_available, not(test)))]
517 for (name, ble_config) in ble_instances {
518 let transport_id = self.allocate_transport_id();
519 let adapter = ble_config.adapter().to_string();
520 let mtu = ble_config.mtu();
521 match crate::transport::ble::io::BluerIo::new(&adapter, mtu).await {
522 Ok(io) => {
523 let mut ble = crate::transport::ble::BleTransport::new(
524 transport_id,
525 name,
526 ble_config,
527 io,
528 packet_tx.clone(),
529 );
530 ble.set_local_pubkey(self.identity.pubkey().serialize());
531 transports.push(TransportHandle::Ble(ble));
532 }
533 Err(e) => {
534 tracing::warn!(adapter = %adapter, error = %e, "failed to initialize BLE adapter");
535 }
536 }
537 }
538
539 #[cfg(any(not(bluer_available), test))]
540 if !ble_instances.is_empty() {
541 #[cfg(not(test))]
542 tracing::warn!("BLE transport configured but this build lacks BlueZ support");
543 }
544 }
545
546 transports
547 }
548
549 pub(super) fn find_transport_for_type(&self, transport_type: &str) -> Option<TransportId> {
559 self.transports
560 .iter()
561 .filter(|(id, handle)| {
562 handle.transport_type().name == transport_type
563 && handle.is_operational()
564 && !self.bootstrap_transports.contains(id)
565 })
566 .min_by_key(|(id, _)| id.as_u32())
567 .map(|(id, _)| *id)
568 }
569
570 #[allow(unused_variables)]
576 pub(super) fn resolve_ethernet_addr(
577 &self,
578 addr_str: &str,
579 ) -> Result<(TransportId, TransportAddr), NodeError> {
580 #[cfg(any(target_os = "linux", target_os = "macos"))]
581 {
582 let (iface, mac_str) = addr_str.split_once('/').ok_or_else(|| {
583 NodeError::NoTransportForType(format!(
584 "invalid Ethernet address format '{}': expected 'interface/mac'",
585 addr_str
586 ))
587 })?;
588
589 let transport_id = self
591 .transports
592 .iter()
593 .find(|(_, handle)| {
594 handle.transport_type().name == "ethernet"
595 && handle.is_operational()
596 && handle.interface_name() == Some(iface)
597 })
598 .map(|(id, _)| *id)
599 .ok_or_else(|| {
600 NodeError::NoTransportForType(format!(
601 "no operational Ethernet transport for interface '{}'",
602 iface
603 ))
604 })?;
605
606 let mac = crate::transport::ethernet::parse_mac_string(mac_str).map_err(|e| {
607 NodeError::NoTransportForType(format!("invalid MAC in '{}': {}", addr_str, e))
608 })?;
609
610 Ok((transport_id, TransportAddr::from_bytes(&mac)))
611 }
612 #[cfg(not(any(target_os = "linux", target_os = "macos")))]
613 {
614 Err(NodeError::NoTransportForType(
615 "Ethernet transport is not supported on this platform".to_string(),
616 ))
617 }
618 }
619
620 #[cfg(bluer_available)]
624 pub(super) fn resolve_ble_addr(
625 &self,
626 addr_str: &str,
627 ) -> Result<(TransportId, TransportAddr), NodeError> {
628 let ta = TransportAddr::from_string(addr_str);
629 let adapter = crate::transport::ble::addr::adapter_from_addr(&ta).ok_or_else(|| {
630 NodeError::NoTransportForType(format!(
631 "invalid BLE address format '{}': expected 'adapter/mac'",
632 addr_str
633 ))
634 })?;
635
636 let transport_id = self
638 .transports
639 .iter()
640 .find(|(_, handle)| handle.transport_type().name == "ble" && handle.is_operational())
641 .map(|(id, _)| *id)
642 .ok_or_else(|| {
643 NodeError::NoTransportForType(format!(
644 "no operational BLE transport for adapter '{}'",
645 adapter
646 ))
647 })?;
648
649 crate::transport::ble::addr::BleAddr::parse(addr_str).map_err(|e| {
651 NodeError::NoTransportForType(format!("invalid BLE address '{}': {}", addr_str, e))
652 })?;
653
654 Ok((transport_id, TransportAddr::from_string(addr_str)))
655 }
656}