Skip to main content

rns_net/
driver.rs

1//! Driver loop: receives events, drives the TransportEngine, dispatches actions.
2
3use std::collections::{HashMap, HashSet};
4use std::io;
5use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
6use std::sync::{Arc, Mutex};
7use std::time::{Duration, Instant};
8
9type MemoryRollup = (f64, f64, f64, f64, f64, f64, f64, f64);
10#[cfg(feature = "iface-backbone")]
11type BackboneCandidateOrder = (
12    std::cmp::Reverse<u8>,
13    u8,
14    u8,
15    u8,
16    std::cmp::Reverse<u32>,
17    std::cmp::Reverse<u64>,
18    [u8; 32],
19);
20
21use rns_core::packet::RawPacket;
22use rns_core::transport::announce_verify_queue::{AnnounceVerifyQueue, OverflowPolicy};
23use rns_core::transport::tables::PathEntry;
24use rns_core::transport::types::{InterfaceId, ReceivedAnnounce, TransportAction, TransportConfig};
25use rns_core::transport::TransportEngine;
26use rns_crypto::{OsRng, Rng};
27
28#[cfg(feature = "hooks")]
29use crate::provider_bridge::ProviderBridge;
30#[cfg(feature = "hooks")]
31use rns_hooks::{create_hook_slots, EngineAccess, HookContext, HookManager, HookPoint, HookSlot};
32
33#[cfg(feature = "hooks")]
34use crate::event::BackbonePeerHookEvent;
35use crate::event::{
36    BackbonePeerPoolMemberStatus, BackbonePeerPoolStatus, BackbonePeerStateEntry, BlackholeInfo,
37    DrainStatus, Event, EventReceiver, InterfaceStatsResponse, KnownDestinationEntry,
38    LifecycleState, LocalDestinationEntry, NextHopResponse, PathTableEntry, QueryRequest,
39    QueryResponse, RateTableEntry, RuntimeConfigApplyMode, RuntimeConfigEntry, RuntimeConfigError,
40    RuntimeConfigErrorCode, RuntimeConfigSource, RuntimeConfigValue, SingleInterfaceStat,
41    TrafficDetail,
42};
43use crate::holepunch::orchestrator::{HolePunchManager, HolePunchManagerAction};
44use crate::ifac;
45#[cfg(all(feature = "iface-auto", test))]
46use crate::interface::auto::AutoRuntime;
47#[cfg(feature = "iface-auto")]
48use crate::interface::auto::AutoRuntimeConfigHandle;
49#[cfg(feature = "iface-backbone")]
50use crate::interface::backbone::{
51    start_client_with_ifac, BackboneClientConfig, BackboneClientRuntime,
52    BackboneClientRuntimeConfigHandle, BackbonePeerStateHandle, BackboneRuntimeConfigHandle,
53};
54#[cfg(all(feature = "iface-backbone", target_os = "linux", test))]
55use crate::interface::backbone::{BackboneAbuseConfig, BackboneServerRuntime};
56#[cfg(all(feature = "iface-i2p", test))]
57use crate::interface::i2p::I2pRuntime;
58#[cfg(feature = "iface-i2p")]
59use crate::interface::i2p::I2pRuntimeConfigHandle;
60#[cfg(all(feature = "iface-pipe", test))]
61use crate::interface::pipe::PipeRuntime;
62#[cfg(feature = "iface-pipe")]
63use crate::interface::pipe::PipeRuntimeConfigHandle;
64#[cfg(feature = "iface-rnode")]
65use crate::interface::rnode::{
66    validate_sub_config, RNodeRuntime, RNodeRuntimeConfigHandle, RNodeSubConfig,
67};
68#[cfg(feature = "iface-tcp")]
69use crate::interface::tcp::TcpClientRuntimeConfigHandle;
70#[cfg(all(feature = "iface-tcp", test))]
71use crate::interface::tcp_server::TcpServerRuntime;
72#[cfg(feature = "iface-tcp")]
73use crate::interface::tcp_server::TcpServerRuntimeConfigHandle;
74#[cfg(all(feature = "iface-udp", test))]
75use crate::interface::udp::UdpRuntime;
76#[cfg(feature = "iface-udp")]
77use crate::interface::udp::UdpRuntimeConfigHandle;
78use crate::interface::{InterfaceEntry, InterfaceStats};
79use crate::link_manager::{LinkManager, LinkManagerAction};
80use crate::time;
81
82const DEFAULT_KNOWN_DESTINATIONS_TTL: f64 = 48.0 * 60.0 * 60.0;
83const DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES: usize = 8192;
84const DEFAULT_RATE_LIMITER_TTL_SECS: f64 = 48.0 * 60.0 * 60.0;
85const DEFAULT_TICK_INTERVAL_MS: u64 = 1000;
86const DEFAULT_KNOWN_DESTINATIONS_CLEANUP_INTERVAL_TICKS: u32 = 3600;
87const KNOWN_DESTINATION_CLEANUP_BATCH_SIZE: usize = 64;
88const DEFAULT_ANNOUNCE_CACHE_CLEANUP_INTERVAL_TICKS: u32 = 3600;
89const DEFAULT_ANNOUNCE_CACHE_CLEANUP_BATCH_SIZE: usize = 10_000;
90const DEFAULT_DISCOVERY_CLEANUP_INTERVAL_TICKS: u32 = 3600;
91const DEFAULT_MANAGEMENT_ANNOUNCE_INTERVAL_SECS: f64 = 3600.0;
92const DEFAULT_LINK_TEARDOWN_FLUSH: Duration = Duration::from_millis(150);
93const SEND_RETRY_BACKOFF_MIN: Duration = Duration::from_millis(25);
94const SEND_RETRY_BACKOFF_MAX: Duration = Duration::from_millis(1000);
95
96/// Default announce-rate controls applied to transport-node interfaces.
97#[derive(Debug, Clone, Copy, PartialEq)]
98pub struct AnnounceRateDefaults {
99    pub target: Option<f64>,
100    pub penalty: f64,
101    pub grace: u32,
102}
103
104impl Default for AnnounceRateDefaults {
105    fn default() -> Self {
106        Self {
107            target: Some(3600.0),
108            penalty: 0.0,
109            grace: 5,
110        }
111    }
112}
113
114mod dispatch;
115mod events;
116mod interface_management;
117mod lifecycle;
118mod queries;
119mod resource_preparation;
120mod runtime_config;
121
122#[cfg(test)]
123pub(crate) mod tests;
124
125fn inject_transport_header(raw: &[u8], next_hop: &[u8; 16]) -> Vec<u8> {
126    if raw.len() < 18 {
127        return raw.to_vec();
128    }
129
130    let new_flags = (rns_core::constants::HEADER_2 << 6)
131        | (rns_core::constants::TRANSPORT_TRANSPORT << 4)
132        | (raw[0] & 0x0F);
133
134    let mut new_raw = Vec::with_capacity(raw.len() + 16);
135    new_raw.push(new_flags);
136    new_raw.push(raw[1]);
137    new_raw.extend_from_slice(next_hop);
138    new_raw.extend_from_slice(&raw[2..]);
139    new_raw
140}
141
142fn recover_mutex_guard<'a, T>(mutex: &'a Mutex<T>, label: &str) -> std::sync::MutexGuard<'a, T> {
143    match mutex.lock() {
144        Ok(guard) => guard,
145        Err(poisoned) => {
146            log::error!("recovering from poisoned mutex: {}", label);
147            poisoned.into_inner()
148        }
149    }
150}
151
152#[derive(Debug, Clone, Copy)]
153pub(crate) struct RuntimeConfigDefaults {
154    pub(crate) tick_interval_ms: u64,
155    pub(crate) known_destinations_ttl: f64,
156    pub(crate) rate_limiter_ttl_secs: f64,
157    pub(crate) known_destinations_cleanup_interval_ticks: u32,
158    pub(crate) announce_cache_cleanup_interval_ticks: u32,
159    pub(crate) announce_cache_cleanup_batch_size: usize,
160    pub(crate) discovery_cleanup_interval_ticks: u32,
161    pub(crate) management_announce_interval_secs: f64,
162    pub(crate) direct_connect_policy: crate::event::HolePunchPolicy,
163    #[cfg(feature = "hooks")]
164    pub(crate) provider_queue_max_events: usize,
165    #[cfg(feature = "hooks")]
166    pub(crate) provider_queue_max_bytes: usize,
167}
168
169#[cfg(feature = "iface-backbone")]
170#[derive(Debug, Clone)]
171pub(crate) struct BackboneDiscoveryRuntime {
172    pub(crate) discoverable: bool,
173    pub(crate) config: crate::discovery::DiscoveryConfig,
174    pub(crate) transport_enabled: bool,
175    pub(crate) ifac_netname: Option<String>,
176    pub(crate) ifac_netkey: Option<String>,
177}
178
179#[cfg(feature = "iface-backbone")]
180#[derive(Debug, Clone)]
181pub(crate) struct BackboneDiscoveryRuntimeHandle {
182    pub(crate) interface_name: String,
183    pub(crate) current: BackboneDiscoveryRuntime,
184    pub(crate) startup: BackboneDiscoveryRuntime,
185}
186
187#[cfg(feature = "iface-tcp")]
188#[derive(Debug, Clone)]
189pub(crate) struct TcpServerDiscoveryRuntime {
190    pub(crate) discoverable: bool,
191    pub(crate) config: crate::discovery::DiscoveryConfig,
192    pub(crate) transport_enabled: bool,
193    pub(crate) ifac_netname: Option<String>,
194    pub(crate) ifac_netkey: Option<String>,
195}
196
197#[cfg(feature = "iface-tcp")]
198#[derive(Debug, Clone)]
199pub(crate) struct TcpServerDiscoveryRuntimeHandle {
200    pub(crate) interface_name: String,
201    pub(crate) current: TcpServerDiscoveryRuntime,
202    pub(crate) startup: TcpServerDiscoveryRuntime,
203}
204
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub(crate) struct IfacRuntimeConfig {
207    pub(crate) netname: Option<String>,
208    pub(crate) netkey: Option<String>,
209    pub(crate) size: usize,
210}
211
212#[cfg(feature = "iface-backbone")]
213impl BackboneDiscoveryRuntimeHandle {
214    pub(crate) fn from_parts(
215        interface_name: String,
216        startup_config: crate::discovery::DiscoveryConfig,
217        transport_enabled: bool,
218        ifac_netname: Option<String>,
219        ifac_netkey: Option<String>,
220        discoverable: bool,
221    ) -> Self {
222        let startup = BackboneDiscoveryRuntime {
223            discoverable,
224            config: startup_config,
225            transport_enabled,
226            ifac_netname,
227            ifac_netkey,
228        };
229        Self {
230            interface_name,
231            current: startup.clone(),
232            startup,
233        }
234    }
235}
236
237#[cfg(feature = "iface-tcp")]
238impl TcpServerDiscoveryRuntimeHandle {
239    pub(crate) fn from_parts(
240        interface_name: String,
241        startup_config: crate::discovery::DiscoveryConfig,
242        transport_enabled: bool,
243        ifac_netname: Option<String>,
244        ifac_netkey: Option<String>,
245        discoverable: bool,
246    ) -> Self {
247        let startup = TcpServerDiscoveryRuntime {
248            discoverable,
249            config: startup_config,
250            transport_enabled,
251            ifac_netname,
252            ifac_netkey,
253        };
254        Self {
255            interface_name,
256            current: startup.clone(),
257            startup,
258        }
259    }
260}
261
262impl IfacRuntimeConfig {
263    pub(crate) fn from_parts(netname: Option<String>, netkey: Option<String>, size: usize) -> Self {
264        Self {
265            netname,
266            netkey,
267            size,
268        }
269    }
270}
271
272#[cfg(feature = "iface-backbone")]
273#[derive(Debug, Clone)]
274pub struct BackbonePeerPoolSettings {
275    pub max_connected: usize,
276    pub failure_threshold: usize,
277    pub failure_window: Duration,
278    pub cooldown: Duration,
279}
280
281#[cfg(feature = "iface-backbone")]
282pub(crate) const BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY: u8 = 60;
283#[cfg(feature = "iface-backbone")]
284pub(crate) const BACKBONE_PEER_POOL_DISCOVERED_PRIORITY: u8 = 40;
285
286#[cfg(feature = "iface-backbone")]
287pub(crate) struct BackbonePeerPoolCandidateConfig {
288    pub(crate) client: BackboneClientConfig,
289    pub(crate) mode: u8,
290    pub(crate) gravity: i64,
291    pub(crate) recursive_prs: bool,
292    pub(crate) announces_from_internal: bool,
293    pub(crate) announces_to_internal: Option<bool>,
294    pub(crate) ingress_control: rns_core::transport::types::IngressControlConfig,
295    pub(crate) ifac_runtime: IfacRuntimeConfig,
296    pub(crate) ifac_enabled: bool,
297    pub(crate) interface_type_name: String,
298    pub(crate) source: BackbonePeerPoolCandidateSource,
299    pub(crate) priority: u8,
300    pub(crate) discovery: Option<BackbonePeerPoolDiscoveryCandidate>,
301}
302
303#[cfg(feature = "iface-backbone")]
304#[derive(Debug, Clone, Copy, PartialEq, Eq)]
305pub(crate) enum BackbonePeerPoolCandidateSource {
306    Configured,
307    Discovered,
308}
309
310#[cfg(feature = "iface-backbone")]
311impl BackbonePeerPoolCandidateSource {
312    fn as_str(self) -> &'static str {
313        match self {
314            Self::Configured => "configured",
315            Self::Discovered => "discovered",
316        }
317    }
318}
319
320#[cfg(feature = "iface-backbone")]
321#[derive(Debug, Clone)]
322pub(crate) struct BackbonePeerPoolDiscoveryCandidate {
323    pub(crate) discovery_hash: [u8; 32],
324    pub(crate) status: crate::discovery::DiscoveredStatus,
325    pub(crate) hops: u8,
326    pub(crate) stamp_value: u32,
327    pub(crate) last_heard: f64,
328}
329
330#[cfg(feature = "iface-backbone")]
331struct BackbonePeerPool {
332    settings: BackbonePeerPoolSettings,
333    candidates: Vec<BackbonePeerPoolCandidate>,
334}
335
336#[cfg(feature = "iface-backbone")]
337struct BackbonePeerPoolCandidate {
338    config: BackbonePeerPoolCandidateConfig,
339    active_id: Option<InterfaceId>,
340    failures: Vec<f64>,
341    /// Failed attempts since the last successful connect; drives the
342    /// reconnect backoff of configured peers.
343    consecutive_failures: u32,
344    retry_after: Option<f64>,
345    cooldown_until: Option<f64>,
346    last_error: Option<String>,
347    /// Stops the active connection's reader when the pool evicts it.
348    control: Option<crate::interface::ListenerControl>,
349}
350
351/// Thin wrapper providing `EngineAccess` for a `TransportEngine` + Driver interfaces.
352#[cfg(feature = "hooks")]
353struct EngineRef<'a> {
354    engine: &'a TransportEngine,
355    interfaces: &'a crate::interface::InterfaceMap<InterfaceEntry>,
356    link_manager: &'a LinkManager,
357    now: f64,
358}
359
360#[cfg(feature = "hooks")]
361impl<'a> EngineAccess for EngineRef<'a> {
362    fn has_path(&self, dest: &[u8; 16]) -> bool {
363        self.engine.has_path(dest)
364    }
365    fn hops_to(&self, dest: &[u8; 16]) -> Option<u8> {
366        self.engine.hops_to(dest)
367    }
368    fn next_hop(&self, dest: &[u8; 16]) -> Option<[u8; 16]> {
369        self.engine.next_hop(dest)
370    }
371    fn is_blackholed(&self, identity: &[u8; 16]) -> bool {
372        self.engine.is_blackholed(identity, self.now)
373    }
374    fn interface_name(&self, id: u64) -> Option<String> {
375        self.interfaces
376            .get(&InterfaceId(id))
377            .map(|e| e.info.name.clone())
378    }
379    fn interface_mode(&self, id: u64) -> Option<u8> {
380        self.interfaces.get(&InterfaceId(id)).map(|e| e.info.mode)
381    }
382    fn identity_hash(&self) -> Option<[u8; 16]> {
383        self.engine.identity_hash().copied()
384    }
385    fn announce_rate(&self, id: u64) -> Option<i32> {
386        self.interfaces
387            .get(&InterfaceId(id))
388            .map(|e| (e.stats.outgoing_announce_freq() * 1000.0) as i32)
389    }
390    fn link_state(&self, link_hash: &[u8; 16]) -> Option<u8> {
391        use rns_core::link::types::LinkState;
392        self.link_manager.link_state(link_hash).map(|s| match s {
393            LinkState::Pending => 0,
394            LinkState::Handshake => 1,
395            LinkState::Active => 2,
396            LinkState::Stale => 3,
397            LinkState::Closed => 4,
398        })
399    }
400}
401
402/// Extract the 16-byte destination hash from a raw packet header.
403///
404/// HEADER_1 (raw[0] & 0x40 == 0): dest at bytes 2..18
405/// HEADER_2 (raw[0] & 0x40 != 0): dest at bytes 18..34 (after transport ID)
406#[cfg(any(test, feature = "hooks"))]
407fn extract_dest_hash(raw: &[u8]) -> [u8; 16] {
408    let mut dest = [0u8; 16];
409    if raw.is_empty() {
410        return dest;
411    }
412    let is_header2 = raw[0] & 0x40 != 0;
413    let start = if is_header2 { 18 } else { 2 };
414    let end = start + 16;
415    if raw.len() >= end {
416        dest.copy_from_slice(&raw[start..end]);
417    }
418    dest
419}
420
421/// Execute a hook chain on disjoint Driver fields (avoids &mut self borrow conflict).
422#[cfg(feature = "hooks")]
423fn run_hook_inner(
424    programs: &mut [rns_hooks::LoadedProgram],
425    hook_manager: &Option<HookManager>,
426    engine_access: &dyn EngineAccess,
427    ctx: &HookContext,
428    now: f64,
429    provider_events_enabled: bool,
430) -> Option<rns_hooks::ExecuteResult> {
431    if programs.is_empty() {
432        return None;
433    }
434    let mgr = hook_manager.as_ref()?;
435    mgr.run_chain_with_provider_events(programs, ctx, engine_access, now, provider_events_enabled)
436}
437
438/// What a packet hook point does after its hook chain ran.
439///
440/// Only hook points that hold the raw packet bytes before acting on them apply
441/// a `Modify` verdict (`PreIngress`, `SendOnInterface`,
442/// `BroadcastOnAllInterfaces`, `DeliverLocal`). Every other hook point treats
443/// `Modify` as `Continue` and ignores the replacement bytes.
444#[cfg(feature = "hooks")]
445pub(crate) enum PacketHookOutcome {
446    /// Carry on with the bytes the hook point already holds.
447    Keep,
448    /// Stop processing this packet.
449    Drop,
450    /// Carry on with the replacement bytes, decoded with the same decoder the
451    /// normal path uses. Hashes and header fields come from the replacement.
452    Replace(RawPacket),
453}
454
455/// Validate replacement bytes from a `Modify` verdict: they must be non-empty,
456/// fit within `max_len`, and decode as a packet.
457#[cfg(feature = "hooks")]
458fn validate_hook_replacement(data: &[u8], max_len: usize) -> Result<RawPacket, &'static str> {
459    if data.is_empty() {
460        return Err("replacement is empty");
461    }
462    if data.len() > max_len {
463        return Err("replacement exceeds the packet size limit");
464    }
465    RawPacket::unpack(data).map_err(|_| "replacement does not decode as a packet")
466}
467
468#[cfg(feature = "hooks")]
469fn backbone_peer_hook_context(event: &BackbonePeerHookEvent) -> HookContext<'_> {
470    HookContext::BackbonePeer {
471        server_interface_id: event.server_interface_id.0,
472        peer_interface_id: event.peer_interface_id.map(|id| id.0),
473        peer_ip: event.peer_ip,
474        peer_port: event.peer_port,
475        connected_for: event.connected_for,
476        had_received_data: event.had_received_data,
477        penalty_level: event.penalty_level,
478        blacklist_for: event.blacklist_for,
479    }
480}
481
482/// Convert a Vec of ActionWire into TransportActions for dispatch.
483#[cfg(feature = "hooks")]
484fn convert_injected_actions(actions: Vec<rns_hooks::ActionWire>) -> Vec<TransportAction> {
485    actions
486        .into_iter()
487        .map(|a| {
488            use rns_hooks::ActionWire;
489            match a {
490                ActionWire::SendOnInterface { interface, raw } => {
491                    TransportAction::SendOnInterface {
492                        interface: InterfaceId(interface),
493                        raw: raw.into(),
494                    }
495                }
496                ActionWire::BroadcastOnAllInterfaces {
497                    raw,
498                    exclude,
499                    has_exclude,
500                } => TransportAction::BroadcastOnAllInterfaces {
501                    raw: raw.into(),
502                    exclude: if has_exclude != 0 {
503                        Some(InterfaceId(exclude))
504                    } else {
505                        None
506                    },
507                },
508                ActionWire::DeliverLocal {
509                    destination_hash,
510                    raw,
511                    packet_hash,
512                    receiving_interface,
513                } => TransportAction::DeliverLocal {
514                    destination_hash,
515                    raw: raw.into(),
516                    packet_hash,
517                    receiving_interface: InterfaceId(receiving_interface),
518                },
519                ActionWire::PathUpdated {
520                    destination_hash,
521                    hops,
522                    next_hop,
523                    interface,
524                } => TransportAction::PathUpdated {
525                    destination_hash,
526                    hops,
527                    next_hop,
528                    interface: InterfaceId(interface),
529                },
530                ActionWire::CacheAnnounce { packet_hash, raw } => TransportAction::CacheAnnounce {
531                    packet_hash,
532                    raw: raw.into(),
533                },
534                ActionWire::TunnelEstablished {
535                    tunnel_id,
536                    interface,
537                } => TransportAction::TunnelEstablished {
538                    tunnel_id,
539                    interface: InterfaceId(interface),
540                },
541                ActionWire::TunnelSynthesize {
542                    interface,
543                    data,
544                    dest_hash,
545                } => TransportAction::TunnelSynthesize {
546                    interface: InterfaceId(interface),
547                    data,
548                    dest_hash,
549                },
550                ActionWire::ForwardToLocalClients {
551                    raw,
552                    exclude,
553                    has_exclude,
554                } => TransportAction::ForwardToLocalClients {
555                    raw: raw.into(),
556                    exclude: if has_exclude != 0 {
557                        Some(InterfaceId(exclude))
558                    } else {
559                        None
560                    },
561                },
562                ActionWire::ForwardPlainBroadcast {
563                    raw,
564                    to_local,
565                    exclude,
566                    has_exclude,
567                } => TransportAction::ForwardPlainBroadcast {
568                    raw: raw.into(),
569                    to_local: to_local != 0,
570                    exclude: if has_exclude != 0 {
571                        Some(InterfaceId(exclude))
572                    } else {
573                        None
574                    },
575                },
576                ActionWire::AnnounceReceived {
577                    destination_hash,
578                    identity_hash,
579                    public_key,
580                    name_hash,
581                    random_hash,
582                    app_data,
583                    hops,
584                    receiving_interface,
585                } => TransportAction::AnnounceReceived(Box::new(ReceivedAnnounce {
586                    destination_hash,
587                    identity_hash,
588                    public_key,
589                    name_hash,
590                    random_hash,
591                    ratchet: None,
592                    app_data,
593                    hops,
594                    receiving_interface: InterfaceId(receiving_interface),
595                    rx: rns_core::transport::RxMetadata::default(),
596                })),
597            }
598        })
599        .collect()
600}
601
602/// Infer the interface type string from a dynamic interface's name.
603/// Dynamic interfaces (TCP server clients, backbone peers, auto peers, local server clients)
604/// include their type in the name prefix set at construction.
605fn infer_interface_type(name: &str) -> String {
606    if name.starts_with("TCPServerInterface") {
607        "TCPServerClientInterface".to_string()
608    } else if name.starts_with("BackboneInterface") {
609        "BackboneInterface".to_string()
610    } else if name.starts_with("LocalInterface") {
611        "LocalServerClientInterface".to_string()
612    } else if name.starts_with("WeaveInterfacePeer[") {
613        "WeaveInterfacePeer".to_string()
614    } else {
615        // AutoInterface peers use "{group_name}:{peer_addr}" format where
616        // group_name is the config section name (typically "AutoInterface" or similar).
617        "AutoInterface".to_string()
618    }
619}
620
621pub use crate::common::callbacks::Callbacks;
622
623#[derive(Clone)]
624struct SharedAnnounceRecord {
625    name_hash: [u8; 10],
626    identity_prv_key: [u8; 64],
627    app_data: Option<Vec<u8>>,
628}
629
630/// Cache of path-response announces for local destinations:
631/// (destination, path-request tag) → (created, raw announce).
632type LocalPathResponseCache = HashMap<([u8; 16], Vec<u8>), (f64, Vec<u8>)>;
633
634#[derive(Debug, Clone)]
635pub(crate) struct KnownDestinationState {
636    announced: crate::destination::AnnouncedIdentity,
637    was_used: bool,
638    last_used_at: Option<f64>,
639    retained: bool,
640}
641
642struct KnownDestinationCleanup {
643    candidates: Vec<[u8; 16]>,
644    cursor: usize,
645    started_at: Instant,
646    removed: usize,
647}
648
649#[derive(Debug, Clone, Copy)]
650pub(crate) struct TrafficSample {
651    sampled_at: f64,
652    rxb: u64,
653    txb: u64,
654    rx_packets: u64,
655    tx_packets: u64,
656    arxb: u64,
657    atxb: u64,
658    prxb: u64,
659    ptxb: u64,
660}
661
662impl TrafficSample {
663    fn from_stats(sampled_at: f64, stats: &crate::interface::InterfaceStats) -> Self {
664        Self {
665            sampled_at,
666            rxb: stats.rxb,
667            txb: stats.txb,
668            rx_packets: stats.rx_packets,
669            tx_packets: stats.tx_packets,
670            arxb: stats.arxb,
671            atxb: stats.atxb,
672            prxb: stats.prxb,
673            ptxb: stats.ptxb,
674        }
675    }
676}
677
678/// The driver loop. Owns the engine and all interface entries.
679pub(crate) struct ManagedInterface {
680    pub(crate) parent_id: InterfaceId,
681    pub(crate) type_name: String,
682    pub(crate) control: Option<crate::interface::ListenerControl>,
683    pub(crate) static_ids: Vec<InterfaceId>,
684}
685
686pub struct Driver {
687    resource_preparation: Option<resource_preparation::Worker>,
688    pub(crate) tracked_link_send: Option<([u8; 32], crate::link_send::Completion)>,
689    pub(crate) pending_link_frames: std::collections::VecDeque<PendingLinkFrame>,
690    pub(crate) engine: TransportEngine,
691    pub(crate) interfaces: crate::interface::InterfaceMap<InterfaceEntry>,
692    /// Parent listener for dynamically spawned interfaces.
693    pub(crate) dynamic_interface_parents: HashMap<InterfaceId, InterfaceId>,
694    /// Previous cumulative counters used to calculate current one-second rates.
695    pub(crate) traffic_samples: HashMap<InterfaceId, TrafficSample>,
696    pub(crate) rng: OsRng,
697    pub(crate) rx: EventReceiver,
698    pub(crate) callbacks: Box<dyn Callbacks>,
699    pub(crate) started: f64,
700    pub(crate) lifecycle_state: LifecycleState,
701    /// Whether `lifecycle_state` is still `Active`, shared with the node
702    /// handle so that each API call can check it without a driver query.
703    pub(crate) accepting_work: Arc<AtomicBool>,
704    pub(crate) drain_started_at: Option<Instant>,
705    pub(crate) drain_deadline: Option<Instant>,
706    pub(crate) listener_controls: Vec<crate::interface::ListenerControl>,
707    pub(crate) interface_management: Option<crate::node::InterfaceManagementConfig>,
708    pub(crate) managed_interfaces: HashMap<String, ManagedInterface>,
709    pub(crate) retired_interface_parents: HashSet<InterfaceId>,
710    /// Writers that may hold frames queued without a wakeup.
711    pub(crate) writers_to_wake: Vec<InterfaceId>,
712    /// Ingress burst state last published to the event queue, per interface.
713    pub(crate) published_ingress_bursts: crate::interface::InterfaceMap<(Option<f64>, Option<f64>)>,
714    pub(crate) announce_cache: Option<crate::announce_cache::AnnounceCache>,
715    /// Destination hash for rnstransport.tunnel.synthesize (PLAIN).
716    pub(crate) tunnel_synth_dest: [u8; 16],
717    /// Transport identity (optional, needed for tunnel synthesis).
718    pub(crate) transport_identity: Option<rns_crypto::identity::Identity>,
719    /// Link manager: handles link lifecycle, request/response.
720    pub(crate) link_manager: LinkManager,
721    /// Management configuration for ACL checks.
722    pub(crate) management_config: crate::management::ManagementConfig,
723    /// Last time management announces were emitted.
724    pub(crate) last_management_announce: f64,
725    /// Whether initial management announce has been sent (delayed 5s after start).
726    pub(crate) initial_announce_sent: bool,
727    /// Cache of known announced identities and lifecycle state, keyed by destination hash.
728    pub(crate) known_destinations: HashMap<[u8; 16], KnownDestinationState>,
729    /// Store for received remote ratchets, if persistence/use is enabled.
730    pub(crate) ratchet_store: Option<Arc<dyn crate::storage::RatchetStore>>,
731    /// TTL for received ratchets, in seconds.
732    pub(crate) ratchet_expiry_secs: f64,
733    /// TTL for known destinations without an active path, in seconds.
734    pub(crate) known_destinations_ttl: f64,
735    /// Maximum number of retained known destinations.
736    pub(crate) known_destinations_max_entries: usize,
737    /// TTL for announce rate-limiter entries without an active path, in seconds.
738    pub(crate) rate_limiter_ttl_secs: f64,
739    /// Destination hash for rnstransport.path.request (PLAIN).
740    pub(crate) path_request_dest: [u8; 16],
741    /// Proof strategies per destination hash.
742    /// Maps dest_hash → (strategy, optional signing identity for generating proofs).
743    pub(crate) proof_strategies: HashMap<
744        [u8; 16],
745        (
746            rns_core::types::ProofStrategy,
747            Option<rns_crypto::identity::Identity>,
748        ),
749    >,
750    /// Tracked sent packets for proof matching: packet_hash → (dest_hash, sent_time).
751    pub(crate) sent_packets: HashMap<[u8; 32], ([u8; 16], f64)>,
752    /// Completed proofs for probe polling: packet_hash → (rtt_seconds, received_time).
753    pub(crate) completed_proofs: HashMap<[u8; 32], (f64, f64)>,
754    /// Locally registered destinations: hash → dest_type.
755    pub(crate) local_destinations: HashMap<[u8; 16], u8>,
756    /// Latest explicit SINGLE announces to replay after shared-client reconnect.
757    shared_announces: HashMap<[u8; 16], SharedAnnounceRecord>,
758    /// Shared local interfaces that went down and should replay announces on reconnect.
759    shared_reconnect_pending: HashMap<InterfaceId, bool>,
760    /// Path-response announces recently sent for local destinations, keyed by
761    /// (destination, path-request tag), replayed byte-for-byte for a repeated
762    /// tag within `LOCAL_PATH_RESPONSE_WINDOW`: (created, raw announce).
763    pub(crate) local_path_responses: LocalPathResponseCache,
764    /// Hole-punch manager for direct P2P connections.
765    pub(crate) holepunch_manager: HolePunchManager,
766    /// Event sender for worker threads to send results back to the driver loop.
767    pub(crate) event_tx: crate::event::EventSender,
768    /// Maximum queued outbound frames per interface writer worker.
769    pub(crate) interface_writer_queue_capacity: usize,
770    /// Linux mark inherited by dynamically created IP underlay sockets.
771    pub(crate) underlay_mark: Option<u32>,
772    /// Default announce-rate controls for interfaces on transport nodes.
773    pub(crate) announce_rate_defaults: AnnounceRateDefaults,
774    /// Default ingress/egress controls for interfaces on transport nodes.
775    pub(crate) ingress_control_defaults: rns_core::transport::types::IngressControlConfig,
776    /// Shared timer interval used by the node timer thread.
777    pub(crate) tick_interval_ms: Arc<AtomicU64>,
778    /// Runtime-config handles for backbone server interfaces, keyed by config name.
779    #[cfg(feature = "iface-backbone")]
780    pub(crate) backbone_runtime: HashMap<String, BackboneRuntimeConfigHandle>,
781    /// Live peer-state handles for backbone server interfaces, keyed by config name.
782    #[cfg(feature = "iface-backbone")]
783    pub(crate) backbone_peer_state: HashMap<String, BackbonePeerStateHandle>,
784    /// Runtime-config handles for backbone client interfaces, keyed by config name.
785    #[cfg(feature = "iface-backbone")]
786    pub(crate) backbone_client_runtime: HashMap<String, BackboneClientRuntimeConfigHandle>,
787    /// Runtime-config state for backbone discovery metadata, keyed by config name.
788    #[cfg(feature = "iface-backbone")]
789    pub(crate) backbone_discovery_runtime: HashMap<String, BackboneDiscoveryRuntimeHandle>,
790    /// Ordered outbound Backbone peer pool, if enabled.
791    #[cfg(feature = "iface-backbone")]
792    backbone_peer_pool: Option<BackbonePeerPool>,
793    /// Shared allocator for listener-created and discovery-created dynamic interfaces.
794    pub(crate) next_dynamic_interface_id: Arc<AtomicU64>,
795    /// Runtime-config handles for TCP server interfaces, keyed by config name.
796    #[cfg(feature = "iface-tcp")]
797    pub(crate) tcp_server_runtime: HashMap<String, TcpServerRuntimeConfigHandle>,
798    /// Runtime-config handles for TCP client interfaces, keyed by config name.
799    #[cfg(feature = "iface-tcp")]
800    pub(crate) tcp_client_runtime: HashMap<String, TcpClientRuntimeConfigHandle>,
801    /// Runtime-config state for TCP server discovery metadata, keyed by config name.
802    #[cfg(feature = "iface-tcp")]
803    pub(crate) tcp_server_discovery_runtime: HashMap<String, TcpServerDiscoveryRuntimeHandle>,
804    /// Runtime-config handles for UDP interfaces, keyed by config name.
805    #[cfg(feature = "iface-udp")]
806    pub(crate) udp_runtime: HashMap<String, UdpRuntimeConfigHandle>,
807    /// Runtime-config handles for Auto interfaces, keyed by config name.
808    #[cfg(feature = "iface-auto")]
809    pub(crate) auto_runtime: HashMap<String, AutoRuntimeConfigHandle>,
810    /// Runtime-config handles for I2P interfaces, keyed by config name.
811    #[cfg(feature = "iface-i2p")]
812    pub(crate) i2p_runtime: HashMap<String, I2pRuntimeConfigHandle>,
813    /// Runtime-config handles for Pipe interfaces, keyed by config name.
814    #[cfg(feature = "iface-pipe")]
815    pub(crate) pipe_runtime: HashMap<String, PipeRuntimeConfigHandle>,
816    /// Runtime-config handles for RNode interfaces, keyed by config name.
817    #[cfg(feature = "iface-rnode")]
818    pub(crate) rnode_runtime: HashMap<String, RNodeRuntimeConfigHandle>,
819    /// Startup/default interface metadata for generic cross-cutting runtime config.
820    pub(crate) interface_runtime_defaults:
821        HashMap<String, rns_core::transport::types::InterfaceInfo>,
822    /// Current IFAC runtime config for static interfaces that support IFAC mutation.
823    pub(crate) interface_ifac_runtime: HashMap<String, IfacRuntimeConfig>,
824    /// Startup/default IFAC runtime config for static interfaces.
825    pub(crate) interface_ifac_runtime_defaults: HashMap<String, IfacRuntimeConfig>,
826    /// Storage for discovered interfaces.
827    pub(crate) discovered_interfaces: crate::discovery::DiscoveredInterfaceStorage,
828    /// Required stamp value for accepting discovered interfaces.
829    pub(crate) discovery_required_value: u8,
830    /// Bounded cache of discovery payloads with previously rejected stamps.
831    pub(crate) discovery_stamp_cache: crate::discovery::DiscoveryStampCache,
832    /// Name hash for interface discovery announces ("rnstransport.discovery.interface").
833    pub(crate) discovery_name_hash: [u8; 10],
834    /// Destination hash for the probe responder (if respond_to_probes is enabled).
835    pub(crate) probe_responder_hash: Option<[u8; 16]>,
836    /// Whether interface discovery is enabled.
837    pub(crate) discover_interfaces: bool,
838    /// Explicit mode for interfaces created from discovery announces.
839    pub(crate) autoconnect_interface_mode: Option<u8>,
840    /// Gravity for interfaces created from discovery announces.
841    pub(crate) autoconnect_interface_gravity: i64,
842    /// Internal announce propagation override for discovered interfaces.
843    pub(crate) autoconnect_announces_to_internal: Option<bool>,
844    /// Allow auto-connecting discovered interfaces from unverified implementations.
845    pub(crate) autoconnect_unverified_implementations: bool,
846    /// Announcer for discoverable interfaces (None if nothing to announce).
847    pub(crate) interface_announcer: Option<crate::discovery::InterfaceAnnouncer>,
848    /// Shared async announce verification queue.
849    pub(crate) announce_verify_queue: Arc<Mutex<AnnounceVerifyQueue>>,
850    /// Whether inbound announces should be verified off the driver thread.
851    pub(crate) async_announce_verification: bool,
852    /// Tick counter for periodic discovery cleanup (every ~3600 ticks = ~1 hour).
853    pub(crate) discovery_cleanup_counter: u32,
854    /// Runtime-configurable discovery cleanup interval.
855    pub(crate) discovery_cleanup_interval_ticks: u32,
856    /// Tick counter for periodic MEMSTATS logging (every 300 ticks = ~5 min).
857    pub(crate) memory_stats_counter: u32,
858    /// Tick counter for periodic memory/cache cleanup (every ~3600 ticks = ~1 hour).
859    pub(crate) cache_cleanup_counter: u32,
860    /// Tick counter for incremental announce-cache cleanup scheduling.
861    pub(crate) announce_cache_cleanup_counter: u32,
862    /// Runtime-configurable cleanup interval for known destinations.
863    pub(crate) known_destinations_cleanup_interval_ticks: u32,
864    /// Incremental known-destination cleanup pass, if one is active.
865    known_destination_cleanup: Option<KnownDestinationCleanup>,
866    /// Low-priority ratchet filesystem cleanup associated with the last pass.
867    ratchet_cleanup_handle: Option<std::thread::JoinHandle<()>>,
868    /// Count of known-destination cap evictions since start.
869    pub(crate) known_destinations_cap_evict_count: usize,
870    /// Runtime-configurable interval for starting announce cache cleanup.
871    pub(crate) announce_cache_cleanup_interval_ticks: u32,
872    /// When set, announce cache cleanup is in progress (contains active packet hashes).
873    pub(crate) cache_cleanup_active_hashes: Option<Vec<[u8; 32]>>,
874    /// Directory iterator for incremental announce cache cleanup.
875    pub(crate) cache_cleanup_entries: Option<std::fs::ReadDir>,
876    /// Running total of files removed during current cache cleanup cycle.
877    pub(crate) cache_cleanup_removed: usize,
878    /// Runtime-configurable announce cache cleanup batch size.
879    pub(crate) announce_cache_cleanup_batch_size: usize,
880    /// Runtime-configurable management announce interval.
881    pub(crate) management_announce_interval_secs: f64,
882    /// Startup/default runtime-config values.
883    pub(crate) runtime_config_defaults: RuntimeConfigDefaults,
884    /// Hook slots for the programmable hook system (one per HookPoint).
885    #[cfg(feature = "hooks")]
886    pub(crate) hook_slots: [HookSlot; HookPoint::COUNT],
887    /// Hook manager. None if initialization failed.
888    #[cfg(feature = "hooks")]
889    pub(crate) hook_manager: Option<HookManager>,
890    #[cfg(feature = "hooks")]
891    pub(crate) provider_bridge: Option<ProviderBridge>,
892    /// Per hook point count of `Modify` replacements that failed validation
893    /// and were therefore treated as `Drop`.
894    #[cfg(feature = "hooks")]
895    pub(crate) hook_rejected_modifications: [u64; HookPoint::COUNT],
896}
897
898impl Driver {
899    /// Create a new driver.
900    pub fn new(
901        config: TransportConfig,
902        rx: EventReceiver,
903        tx: crate::event::EventSender,
904        callbacks: Box<dyn Callbacks>,
905    ) -> Self {
906        let announce_queue_max_entries = config.announce_queue_max_entries;
907        let tunnel_synth_dest = rns_core::destination::destination_hash(
908            "rnstransport",
909            &["tunnel", "synthesize"],
910            None,
911        );
912        let path_request_dest =
913            rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
914        let discovery_name_hash = crate::discovery::discovery_name_hash();
915        let mut engine = TransportEngine::new(config);
916        engine.seed_table_hashing(&mut OsRng);
917        engine.register_destination(tunnel_synth_dest, rns_core::constants::DESTINATION_PLAIN);
918        // Register path request destination so inbound path requests are delivered locally
919        engine.register_destination(path_request_dest, rns_core::constants::DESTINATION_PLAIN);
920        // Note: discovery destination is NOT registered as local — it's a SINGLE destination
921        // whose hash depends on the sender's identity. We match it by name_hash instead.
922        let mut local_destinations = HashMap::new();
923        local_destinations.insert(tunnel_synth_dest, rns_core::constants::DESTINATION_PLAIN);
924        local_destinations.insert(path_request_dest, rns_core::constants::DESTINATION_PLAIN);
925        let runtime_config_defaults = RuntimeConfigDefaults {
926            tick_interval_ms: DEFAULT_TICK_INTERVAL_MS,
927            known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
928            rate_limiter_ttl_secs: DEFAULT_RATE_LIMITER_TTL_SECS,
929            known_destinations_cleanup_interval_ticks:
930                DEFAULT_KNOWN_DESTINATIONS_CLEANUP_INTERVAL_TICKS,
931            announce_cache_cleanup_interval_ticks: DEFAULT_ANNOUNCE_CACHE_CLEANUP_INTERVAL_TICKS,
932            announce_cache_cleanup_batch_size: DEFAULT_ANNOUNCE_CACHE_CLEANUP_BATCH_SIZE,
933            discovery_cleanup_interval_ticks: DEFAULT_DISCOVERY_CLEANUP_INTERVAL_TICKS,
934            management_announce_interval_secs: DEFAULT_MANAGEMENT_ANNOUNCE_INTERVAL_SECS,
935            direct_connect_policy: crate::event::HolePunchPolicy::default(),
936            #[cfg(feature = "hooks")]
937            provider_queue_max_events: crate::provider_bridge::ProviderBridgeConfig::default()
938                .queue_max_events,
939            #[cfg(feature = "hooks")]
940            provider_queue_max_bytes: crate::provider_bridge::ProviderBridgeConfig::default()
941                .queue_max_bytes,
942        };
943        Driver {
944            tracked_link_send: None,
945            pending_link_frames: std::collections::VecDeque::new(),
946            engine,
947            interfaces: Default::default(),
948            dynamic_interface_parents: HashMap::new(),
949            traffic_samples: HashMap::new(),
950            rng: OsRng,
951            rx,
952            callbacks,
953            started: time::now(),
954            lifecycle_state: LifecycleState::Active,
955            accepting_work: Arc::new(AtomicBool::new(true)),
956            drain_started_at: None,
957            drain_deadline: None,
958            listener_controls: Vec::new(),
959            interface_management: None,
960            managed_interfaces: HashMap::new(),
961            retired_interface_parents: HashSet::new(),
962            published_ingress_bursts: Default::default(),
963            writers_to_wake: Vec::new(),
964            announce_cache: None,
965            tunnel_synth_dest,
966            transport_identity: None,
967            link_manager: LinkManager::new(),
968            management_config: Default::default(),
969            last_management_announce: 0.0,
970            initial_announce_sent: false,
971            known_destinations: HashMap::new(),
972            ratchet_store: None,
973            ratchet_expiry_secs: rns_core::constants::RATCHET_EXPIRY as f64,
974            known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
975            known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
976            rate_limiter_ttl_secs: DEFAULT_RATE_LIMITER_TTL_SECS,
977            path_request_dest,
978            proof_strategies: HashMap::new(),
979            sent_packets: HashMap::new(),
980            completed_proofs: HashMap::new(),
981            local_destinations,
982            shared_announces: HashMap::new(),
983            shared_reconnect_pending: HashMap::new(),
984            local_path_responses: HashMap::new(),
985            holepunch_manager: HolePunchManager::new(
986                vec![],
987                rns_core::holepunch::ProbeProtocol::Rnsp,
988                None,
989            ),
990            event_tx: tx,
991            interface_writer_queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
992            underlay_mark: None,
993            announce_rate_defaults: AnnounceRateDefaults::default(),
994            ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(),
995            tick_interval_ms: Arc::new(AtomicU64::new(DEFAULT_TICK_INTERVAL_MS)),
996            #[cfg(feature = "iface-backbone")]
997            backbone_runtime: HashMap::new(),
998            #[cfg(feature = "iface-backbone")]
999            backbone_peer_state: HashMap::new(),
1000            #[cfg(feature = "iface-backbone")]
1001            backbone_client_runtime: HashMap::new(),
1002            #[cfg(feature = "iface-backbone")]
1003            backbone_discovery_runtime: HashMap::new(),
1004            #[cfg(feature = "iface-backbone")]
1005            backbone_peer_pool: None,
1006            next_dynamic_interface_id: Arc::new(AtomicU64::new(10000)),
1007            #[cfg(feature = "iface-tcp")]
1008            tcp_server_runtime: HashMap::new(),
1009            #[cfg(feature = "iface-tcp")]
1010            tcp_client_runtime: HashMap::new(),
1011            #[cfg(feature = "iface-tcp")]
1012            tcp_server_discovery_runtime: HashMap::new(),
1013            #[cfg(feature = "iface-udp")]
1014            udp_runtime: HashMap::new(),
1015            #[cfg(feature = "iface-auto")]
1016            auto_runtime: HashMap::new(),
1017            #[cfg(feature = "iface-i2p")]
1018            i2p_runtime: HashMap::new(),
1019            #[cfg(feature = "iface-pipe")]
1020            pipe_runtime: HashMap::new(),
1021            #[cfg(feature = "iface-rnode")]
1022            rnode_runtime: HashMap::new(),
1023            interface_runtime_defaults: HashMap::new(),
1024            interface_ifac_runtime: HashMap::new(),
1025            interface_ifac_runtime_defaults: HashMap::new(),
1026            discovered_interfaces: crate::discovery::DiscoveredInterfaceStorage::new(
1027                std::env::temp_dir().join("rns-discovered-interfaces"),
1028            ),
1029            discovery_required_value: crate::discovery::DEFAULT_STAMP_VALUE,
1030            discovery_stamp_cache: crate::discovery::DiscoveryStampCache::default(),
1031            discovery_name_hash,
1032            probe_responder_hash: None,
1033            discover_interfaces: false,
1034            autoconnect_interface_mode: None,
1035            autoconnect_interface_gravity: 0,
1036            autoconnect_announces_to_internal: None,
1037            autoconnect_unverified_implementations: false,
1038            interface_announcer: None,
1039            announce_verify_queue: Arc::new(Mutex::new(AnnounceVerifyQueue::new(
1040                announce_queue_max_entries,
1041            ))),
1042            async_announce_verification: false,
1043            discovery_cleanup_counter: 0,
1044            discovery_cleanup_interval_ticks: runtime_config_defaults
1045                .discovery_cleanup_interval_ticks,
1046            memory_stats_counter: 0,
1047            cache_cleanup_counter: 0,
1048            announce_cache_cleanup_counter: 0,
1049            known_destinations_cleanup_interval_ticks: runtime_config_defaults
1050                .known_destinations_cleanup_interval_ticks,
1051            known_destination_cleanup: None,
1052            ratchet_cleanup_handle: None,
1053            resource_preparation: None,
1054            known_destinations_cap_evict_count: 0,
1055            announce_cache_cleanup_interval_ticks: runtime_config_defaults
1056                .announce_cache_cleanup_interval_ticks,
1057            cache_cleanup_active_hashes: None,
1058            cache_cleanup_entries: None,
1059            cache_cleanup_removed: 0,
1060            announce_cache_cleanup_batch_size: runtime_config_defaults
1061                .announce_cache_cleanup_batch_size,
1062            management_announce_interval_secs: runtime_config_defaults
1063                .management_announce_interval_secs,
1064            runtime_config_defaults,
1065            #[cfg(feature = "hooks")]
1066            hook_slots: create_hook_slots(),
1067            #[cfg(feature = "hooks")]
1068            hook_manager: HookManager::new().ok(),
1069            #[cfg(feature = "hooks")]
1070            provider_bridge: None,
1071            #[cfg(feature = "hooks")]
1072            hook_rejected_modifications: [0; HookPoint::COUNT],
1073        }
1074    }
1075
1076    pub fn set_announce_verify_queue_config(
1077        &mut self,
1078        max_entries: usize,
1079        max_bytes: usize,
1080        max_stale_secs: f64,
1081        overflow_policy: OverflowPolicy,
1082    ) {
1083        self.announce_verify_queue = Arc::new(Mutex::new(AnnounceVerifyQueue::with_limits(
1084            max_entries,
1085            max_bytes,
1086            max_stale_secs,
1087            overflow_policy,
1088        )));
1089    }
1090
1091    pub fn set_announce_rate_defaults(&mut self, defaults: AnnounceRateDefaults) {
1092        self.announce_rate_defaults = defaults;
1093    }
1094
1095    pub fn set_ingress_control_defaults(
1096        &mut self,
1097        defaults: rns_core::transport::types::IngressControlConfig,
1098    ) {
1099        self.ingress_control_defaults = defaults;
1100    }
1101
1102    pub(crate) fn apply_announce_rate_defaults(
1103        &self,
1104        info: &mut rns_core::transport::types::InterfaceInfo,
1105    ) {
1106        if !self.engine.transport_enabled() || info.announce_rate_target.is_some() {
1107            return;
1108        }
1109
1110        let Some(target) = self.announce_rate_defaults.target else {
1111            return;
1112        };
1113
1114        info.announce_rate_target = Some(target);
1115        if info.announce_rate_grace == 0 {
1116            info.announce_rate_grace = self.announce_rate_defaults.grace;
1117        }
1118        if info.announce_rate_penalty == 0.0 {
1119            info.announce_rate_penalty = self.announce_rate_defaults.penalty;
1120        }
1121    }
1122
1123    pub(crate) fn apply_ingress_control_defaults(
1124        &self,
1125        info: &mut rns_core::transport::types::InterfaceInfo,
1126    ) {
1127        if !self.engine.transport_enabled() {
1128            return;
1129        }
1130
1131        let baseline = if info.ingress_control.enabled {
1132            rns_core::transport::types::IngressControlConfig::enabled()
1133        } else {
1134            rns_core::transport::types::IngressControlConfig::disabled()
1135        };
1136        let defaults = self.ingress_control_defaults;
1137        let ic = &mut info.ingress_control;
1138
1139        if ic.max_held_announces == baseline.max_held_announces {
1140            ic.max_held_announces = defaults.max_held_announces;
1141        }
1142        if ic.burst_hold == baseline.burst_hold {
1143            ic.burst_hold = defaults.burst_hold;
1144        }
1145        if ic.burst_freq_new == baseline.burst_freq_new {
1146            ic.burst_freq_new = defaults.burst_freq_new;
1147        }
1148        if ic.burst_freq == baseline.burst_freq {
1149            ic.burst_freq = defaults.burst_freq;
1150        }
1151        if ic.pr_burst_freq_new == baseline.pr_burst_freq_new {
1152            ic.pr_burst_freq_new = defaults.pr_burst_freq_new;
1153        }
1154        if ic.pr_burst_freq == baseline.pr_burst_freq {
1155            ic.pr_burst_freq = defaults.pr_burst_freq;
1156        }
1157        if ic.new_time == baseline.new_time {
1158            ic.new_time = defaults.new_time;
1159        }
1160        if ic.burst_penalty == baseline.burst_penalty {
1161            ic.burst_penalty = defaults.burst_penalty;
1162        }
1163        if ic.held_release_interval == baseline.held_release_interval {
1164            ic.held_release_interval = defaults.held_release_interval;
1165        }
1166        if ic.egress_enabled == baseline.egress_enabled {
1167            ic.egress_enabled = defaults.egress_enabled;
1168        }
1169        if ic.egress_pr_freq == baseline.egress_pr_freq {
1170            ic.egress_pr_freq = defaults.egress_pr_freq;
1171        }
1172    }
1173
1174    fn wrap_interface_writer(
1175        &self,
1176        interface_id: InterfaceId,
1177        interface_name: &str,
1178        writer: Box<dyn crate::interface::Writer>,
1179    ) -> (
1180        Box<dyn crate::interface::Writer>,
1181        crate::interface::AsyncWriterMetrics,
1182    ) {
1183        crate::interface::wrap_async_writer(
1184            writer,
1185            interface_id,
1186            interface_name,
1187            self.event_tx.clone(),
1188            self.interface_writer_queue_capacity,
1189        )
1190    }
1191
1192    #[cfg(feature = "hooks")]
1193    fn provider_events_enabled(&self) -> bool {
1194        self.provider_bridge.is_some()
1195    }
1196
1197    /// Whether any program is attached at `point`. Per-packet hook points
1198    /// check this before building a hook context, which reads the clock.
1199    #[cfg(feature = "hooks")]
1200    pub(crate) fn hooks_attached(&self, point: HookPoint) -> bool {
1201        !self.hook_slots[point as usize].programs.is_empty()
1202    }
1203
1204    #[cfg(feature = "hooks")]
1205    fn run_backbone_peer_hook(
1206        &mut self,
1207        attach_point: &str,
1208        point: HookPoint,
1209        event: &BackbonePeerHookEvent,
1210    ) {
1211        let ctx = backbone_peer_hook_context(event);
1212        let now = time::now();
1213        let engine_ref = EngineRef {
1214            engine: &self.engine,
1215            interfaces: &self.interfaces,
1216            link_manager: &self.link_manager,
1217            now,
1218        };
1219        let provider_events_enabled = self.provider_events_enabled();
1220        if let Some(ref e) = run_hook_inner(
1221            &mut self.hook_slots[point as usize].programs,
1222            &self.hook_manager,
1223            &engine_ref,
1224            &ctx,
1225            now,
1226            provider_events_enabled,
1227        ) {
1228            self.forward_hook_side_effects(attach_point, e);
1229        }
1230    }
1231
1232    #[cfg(feature = "iface-backbone")]
1233    fn make_discoverable_interface(
1234        runtime: &BackboneDiscoveryRuntimeHandle,
1235    ) -> crate::discovery::DiscoverableInterface {
1236        crate::discovery::DiscoverableInterface {
1237            interface_name: runtime.interface_name.clone(),
1238            config: runtime.current.config.clone(),
1239            transport_enabled: runtime.current.transport_enabled,
1240            ifac_netname: runtime.current.ifac_netname.clone(),
1241            ifac_netkey: runtime.current.ifac_netkey.clone(),
1242        }
1243    }
1244
1245    #[cfg(feature = "iface-backbone")]
1246    fn sync_backbone_discovery_runtime(
1247        &mut self,
1248        interface_name: &str,
1249    ) -> Result<(), RuntimeConfigError> {
1250        let handle = self
1251            .backbone_discovery_runtime
1252            .get(interface_name)
1253            .ok_or(RuntimeConfigError {
1254                code: RuntimeConfigErrorCode::NotFound,
1255                message: format!("backbone interface '{}' not found", interface_name),
1256            })?
1257            .clone();
1258
1259        if handle.current.discoverable {
1260            let iface = Self::make_discoverable_interface(&handle);
1261            if let Some(announcer) = self.interface_announcer.as_mut() {
1262                announcer.upsert_interface(iface);
1263            } else if let Some(identity) = self.transport_identity.as_ref() {
1264                self.interface_announcer = Some(crate::discovery::InterfaceAnnouncer::new(
1265                    *identity.hash(),
1266                    vec![iface],
1267                ));
1268            }
1269        } else if let Some(announcer) = self.interface_announcer.as_mut() {
1270            announcer.remove_interface(interface_name);
1271            if announcer.is_empty() {
1272                self.interface_announcer = None;
1273            }
1274        }
1275
1276        Ok(())
1277    }
1278
1279    #[cfg(feature = "iface-tcp")]
1280    fn make_tcp_server_discoverable_interface(
1281        runtime: &TcpServerDiscoveryRuntimeHandle,
1282    ) -> crate::discovery::DiscoverableInterface {
1283        crate::discovery::DiscoverableInterface {
1284            interface_name: runtime.interface_name.clone(),
1285            config: runtime.current.config.clone(),
1286            transport_enabled: runtime.current.transport_enabled,
1287            ifac_netname: runtime.current.ifac_netname.clone(),
1288            ifac_netkey: runtime.current.ifac_netkey.clone(),
1289        }
1290    }
1291
1292    #[cfg(feature = "iface-tcp")]
1293    fn sync_tcp_server_discovery_runtime(
1294        &mut self,
1295        interface_name: &str,
1296    ) -> Result<(), RuntimeConfigError> {
1297        let handle = self
1298            .tcp_server_discovery_runtime
1299            .get(interface_name)
1300            .ok_or(RuntimeConfigError {
1301                code: RuntimeConfigErrorCode::NotFound,
1302                message: format!("tcp server interface '{}' not found", interface_name),
1303            })?
1304            .clone();
1305
1306        if handle.current.discoverable {
1307            let iface = Self::make_tcp_server_discoverable_interface(&handle);
1308            if let Some(announcer) = self.interface_announcer.as_mut() {
1309                announcer.upsert_interface(iface);
1310            } else if let Some(identity) = self.transport_identity.as_ref() {
1311                self.interface_announcer = Some(crate::discovery::InterfaceAnnouncer::new(
1312                    *identity.hash(),
1313                    vec![iface],
1314                ));
1315            }
1316        } else if let Some(announcer) = self.interface_announcer.as_mut() {
1317            announcer.remove_interface(interface_name);
1318            if announcer.is_empty() {
1319                self.interface_announcer = None;
1320            }
1321        }
1322
1323        Ok(())
1324    }
1325
1326    #[cfg(feature = "hooks")]
1327    fn update_hook_program<F>(
1328        &mut self,
1329        name: &str,
1330        attach_point: &str,
1331        mut update: F,
1332    ) -> Result<(), String>
1333    where
1334        F: FnMut(&mut rns_hooks::LoadedProgram),
1335    {
1336        let point_idx = crate::config::parse_hook_point(attach_point)
1337            .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
1338        let program = self.hook_slots[point_idx]
1339            .programs
1340            .iter_mut()
1341            .find(|program| program.name == name)
1342            .ok_or_else(|| format!("hook '{}' not found at point '{}'", name, attach_point))?;
1343        update(program);
1344        Ok(())
1345    }
1346
1347    pub(crate) fn set_tick_interval_handle(&mut self, tick_interval_ms: Arc<AtomicU64>) {
1348        self.tick_interval_ms = tick_interval_ms;
1349    }
1350
1351    /// The flag the node handle reads to refuse new work once draining starts.
1352    pub(crate) fn accepting_work_handle(&self) -> Arc<AtomicBool> {
1353        Arc::clone(&self.accepting_work)
1354    }
1355
1356    #[allow(dead_code)]
1357    pub(crate) fn set_packet_hashlist_max_entries(&mut self, max_entries: usize) {
1358        self.engine.set_packet_hashlist_max_entries(max_entries);
1359    }
1360
1361    /// Turn a packet hook chain result into the action the hook point takes.
1362    ///
1363    /// `Drop` stops the packet. `Modify` with replacement bytes is validated
1364    /// against `max_len` and the packet decoder; a replacement that fails
1365    /// validation is treated as `Drop` and counted in
1366    /// `hook_rejected_modifications`. `Modify` without bytes, `Continue` and
1367    /// `Halt` keep the current bytes.
1368    #[cfg(feature = "hooks")]
1369    pub(crate) fn packet_hook_outcome(
1370        &mut self,
1371        point: HookPoint,
1372        exec: &rns_hooks::ExecuteResult,
1373        max_len: usize,
1374    ) -> PacketHookOutcome {
1375        let Some(result) = exec.hook_result.as_ref() else {
1376            return PacketHookOutcome::Keep;
1377        };
1378        if result.is_drop() {
1379            return PacketHookOutcome::Drop;
1380        }
1381        if rns_hooks::Verdict::from_u32(result.verdict) != Some(rns_hooks::Verdict::Modify) {
1382            return PacketHookOutcome::Keep;
1383        }
1384        let Some(data) = exec.modified_data.as_deref() else {
1385            return PacketHookOutcome::Keep;
1386        };
1387        match validate_hook_replacement(data, max_len) {
1388            Ok(packet) => PacketHookOutcome::Replace(packet),
1389            Err(reason) => {
1390                let rejected = &mut self.hook_rejected_modifications[point as usize];
1391                *rejected = rejected.saturating_add(1);
1392                log::debug!("{:?} hook: {}; dropping packet", point, reason);
1393                PacketHookOutcome::Drop
1394            }
1395        }
1396    }
1397
1398    /// Largest packet a hook may substitute for one handled on `interface`.
1399    #[cfg(feature = "hooks")]
1400    pub(crate) fn hook_packet_limit(&self, interface: InterfaceId) -> usize {
1401        self.interfaces
1402            .get(&interface)
1403            .map(|entry| entry.info.mtu as usize)
1404            .filter(|mtu| *mtu > 0)
1405            .unwrap_or(rns_core::constants::MTU)
1406    }
1407
1408    /// Point a pending confirmed link send at the replacement bytes a hook
1409    /// produced for the packet it was tracking.
1410    #[cfg(feature = "hooks")]
1411    pub(crate) fn retarget_tracked_link_send(&mut self, original: &[u8], replacement: &RawPacket) {
1412        let Some((hash, _)) = self.tracked_link_send.as_mut() else {
1413            return;
1414        };
1415        if RawPacket::unpack(original).is_ok_and(|packet| packet.packet_hash == *hash) {
1416            *hash = replacement.packet_hash;
1417        }
1418    }
1419
1420    #[cfg(feature = "hooks")]
1421    fn forward_hook_side_effects(&mut self, attach_point: &str, exec: &rns_hooks::ExecuteResult) {
1422        if !exec.injected_actions.is_empty() {
1423            self.dispatch_all(convert_injected_actions(exec.injected_actions.clone()));
1424        }
1425        if let Some(ref bridge) = self.provider_bridge {
1426            for event in &exec.provider_events {
1427                bridge.emit_event(
1428                    attach_point,
1429                    event.hook_name.clone(),
1430                    event.payload_type.clone(),
1431                    event.payload.clone(),
1432                );
1433            }
1434        }
1435    }
1436
1437    #[cfg(feature = "hooks")]
1438    fn collect_hook_side_effects(
1439        &mut self,
1440        attach_point: &str,
1441        exec: &rns_hooks::ExecuteResult,
1442        out: &mut Vec<TransportAction>,
1443    ) {
1444        if !exec.injected_actions.is_empty() {
1445            out.extend(convert_injected_actions(exec.injected_actions.clone()));
1446        }
1447        if let Some(ref bridge) = self.provider_bridge {
1448            for event in &exec.provider_events {
1449                bridge.emit_event(
1450                    attach_point,
1451                    event.hook_name.clone(),
1452                    event.payload_type.clone(),
1453                    event.payload.clone(),
1454                );
1455            }
1456        }
1457    }
1458
1459    /// Set the probe addresses, protocol, and optional device for hole punching.
1460    pub fn set_probe_config(
1461        &mut self,
1462        addrs: Vec<std::net::SocketAddr>,
1463        protocol: rns_core::holepunch::ProbeProtocol,
1464        device: Option<String>,
1465    ) {
1466        let policy = self.holepunch_manager.policy();
1467        let allow_private = self.holepunch_manager.allow_private_endpoints();
1468        self.holepunch_manager = HolePunchManager::new(addrs, protocol, device);
1469        self.holepunch_manager.set_underlay_mark(self.underlay_mark);
1470        self.holepunch_manager.set_policy(policy);
1471        self.holepunch_manager
1472            .set_allow_private_endpoints(allow_private);
1473    }
1474}
1475
1476pub(crate) struct PendingLinkFrame {
1477    interface: InterfaceId,
1478    data: Vec<u8>,
1479    is_announce: bool,
1480    completion: Option<crate::link_send::Completion>,
1481}