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