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