rns_net/

driver.rs

//! Driver loop: receives events, drives the TransportEngine, dispatches actions.

use std::collections::{HashMap, HashSet};
use std::io;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

type MemoryRollup = (f64, f64, f64, f64, f64, f64, f64, f64);
#[cfg(feature = "iface-backbone")]
type BackboneCandidateOrder = (
    std::cmp::Reverse<u8>,
    u8,
    u8,
    u8,
    std::cmp::Reverse<rns_core::stamp::StampValue>,
    std::cmp::Reverse<u64>,
    [u8; 32],
);

use rns_core::packet::RawPacket;
use rns_core::transport::announce_verify_queue::{AnnounceVerifyQueue, OverflowPolicy};
use rns_core::transport::tables::PathEntry;
use rns_core::transport::types::{InterfaceId, ReceivedAnnounce, TransportAction, TransportConfig};
use rns_core::transport::TransportEngine;
use rns_crypto::{OsRng, Rng};

#[cfg(feature = "hooks")]
use crate::provider_bridge::ProviderBridge;
use crate::stamper::PathRequestTarget;
#[cfg(feature = "hooks")]
use rns_hooks::{create_hook_slots, EngineAccess, HookContext, HookManager, HookPoint, HookSlot};

#[cfg(feature = "hooks")]
use crate::event::BackbonePeerHookEvent;
use crate::event::{
    BackbonePeerPoolMemberStatus, BackbonePeerPoolStatus, BackbonePeerStateEntry, BlackholeInfo,
    DrainStatus, Event, EventReceiver, InterfaceStatsResponse, KnownDestinationEntry,
    LifecycleState, LocalDestinationEntry, NextHopResponse, PathTableEntry, QueryRequest,
    QueryResponse, RateTableEntry, RuntimeConfigApplyMode, RuntimeConfigEntry, RuntimeConfigError,
    RuntimeConfigErrorCode, RuntimeConfigSource, RuntimeConfigValue, SingleInterfaceStat,
    TrafficDetail,
};
use crate::holepunch::orchestrator::{HolePunchManager, HolePunchManagerAction};
use crate::ifac;
#[cfg(all(feature = "iface-auto", test))]
use crate::interface::auto::AutoRuntime;
#[cfg(feature = "iface-auto")]
use crate::interface::auto::AutoRuntimeConfigHandle;
#[cfg(feature = "iface-backbone")]
use crate::interface::backbone::{
    start_client_with_ifac, BackboneClientConfig, BackboneClientRuntime,
    BackboneClientRuntimeConfigHandle, BackbonePeerStateHandle, BackboneRuntimeConfigHandle,
};
#[cfg(all(feature = "iface-backbone", target_os = "linux", test))]
use crate::interface::backbone::{BackboneAbuseConfig, BackboneServerRuntime};
#[cfg(all(feature = "iface-i2p", test))]
use crate::interface::i2p::I2pRuntime;
#[cfg(feature = "iface-i2p")]
use crate::interface::i2p::I2pRuntimeConfigHandle;
#[cfg(all(feature = "iface-pipe", test))]
use crate::interface::pipe::PipeRuntime;
#[cfg(feature = "iface-pipe")]
use crate::interface::pipe::PipeRuntimeConfigHandle;
#[cfg(feature = "iface-rnode")]
use crate::interface::rnode::{
    validate_sub_config, RNodeRuntime, RNodeRuntimeConfigHandle, RNodeSubConfig,
};
#[cfg(feature = "iface-tcp")]
use crate::interface::tcp::TcpClientRuntimeConfigHandle;
#[cfg(all(feature = "iface-tcp", test))]
use crate::interface::tcp_server::TcpServerRuntime;
#[cfg(feature = "iface-tcp")]
use crate::interface::tcp_server::TcpServerRuntimeConfigHandle;
#[cfg(all(feature = "iface-udp", test))]
use crate::interface::udp::UdpRuntime;
#[cfg(feature = "iface-udp")]
use crate::interface::udp::UdpRuntimeConfigHandle;
use crate::interface::{InterfaceEntry, InterfaceStats};
use crate::link_manager::{LinkManager, LinkManagerAction};
use crate::time;

const DEFAULT_KNOWN_DESTINATIONS_TTL: f64 = 48.0 * 60.0 * 60.0;
const DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES: usize = 8192;
const DEFAULT_RATE_LIMITER_TTL_SECS: f64 = 48.0 * 60.0 * 60.0;
const DEFAULT_TICK_INTERVAL_MS: u64 = 1000;
const DEFAULT_KNOWN_DESTINATIONS_CLEANUP_INTERVAL_TICKS: u32 = 3600;
const KNOWN_DESTINATION_CLEANUP_BATCH_SIZE: usize = 64;
const DEFAULT_ANNOUNCE_CACHE_CLEANUP_INTERVAL_TICKS: u32 = 3600;
const DEFAULT_ANNOUNCE_CACHE_CLEANUP_BATCH_SIZE: usize = 10_000;
const DEFAULT_DISCOVERY_CLEANUP_INTERVAL_TICKS: u32 = 3600;
const DEFAULT_MANAGEMENT_ANNOUNCE_INTERVAL_SECS: f64 = 3600.0;
const DEFAULT_LINK_TEARDOWN_FLUSH: Duration = Duration::from_millis(150);
const SEND_RETRY_BACKOFF_MIN: Duration = Duration::from_millis(25);
const SEND_RETRY_BACKOFF_MAX: Duration = Duration::from_millis(1000);

/// Default announce-rate controls applied to transport-node interfaces.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct AnnounceRateDefaults {
    pub target: Option<f64>,
    pub penalty: f64,
    pub grace: u32,
}

impl Default for AnnounceRateDefaults {
    fn default() -> Self {
        Self {
            target: Some(3600.0),
            penalty: 0.0,
            grace: 5,
        }
    }
}

mod announce_check;
pub(crate) use announce_check::{check_pending_announce, AnnounceCheck};
mod dispatch;
mod events;
mod hook_events;
mod interface_management;
mod lifecycle;
mod link_stamps;
mod queries;
mod resource_preparation;
mod rpc_service;
pub(crate) use rpc_service::{RpcListener, RpcService};
mod runtime_config;
#[cfg(any(target_os = "linux", target_os = "android"))]
mod tick_source;

#[cfg(test)]
pub(crate) mod tests;

fn inject_transport_header(raw: &[u8], next_hop: &[u8; 16]) -> Vec<u8> {
    if raw.len() < 18 {
        return raw.to_vec();
    }

    let new_flags = (rns_core::constants::HEADER_2 << 6)
        | (rns_core::constants::TRANSPORT_TRANSPORT << 4)
        | (raw[0] & 0x0F);

    let mut new_raw = Vec::with_capacity(raw.len() + 16);
    new_raw.push(new_flags);
    new_raw.push(raw[1]);
    new_raw.extend_from_slice(next_hop);
    new_raw.extend_from_slice(&raw[2..]);
    new_raw
}

fn recover_mutex_guard<'a, T>(mutex: &'a Mutex<T>, label: &str) -> std::sync::MutexGuard<'a, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(poisoned) => {
            log::error!("recovering from poisoned mutex: {}", label);
            poisoned.into_inner()
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct RuntimeConfigDefaults {
    pub(crate) tick_interval_ms: u64,
    pub(crate) known_destinations_ttl: f64,
    pub(crate) rate_limiter_ttl_secs: f64,
    pub(crate) known_destinations_cleanup_interval_ticks: u32,
    pub(crate) announce_cache_cleanup_interval_ticks: u32,
    pub(crate) announce_cache_cleanup_batch_size: usize,
    pub(crate) discovery_cleanup_interval_ticks: u32,
    pub(crate) management_announce_interval_secs: f64,
    pub(crate) direct_connect_policy: crate::event::HolePunchPolicy,
    /// `Transport.STAMP_DESTINATIONS` default.
    pub(crate) stamp_destinations: bool,
    /// `Transport.STAMP_PRS` default.
    pub(crate) stamp_path_requests: bool,
    /// `Transport.STAMPS_ENFORCED` default.
    pub(crate) stamps_enforced: bool,
    /// `Transport.LR_STAMPS_ENFORCED` default.
    pub(crate) lr_stamps_enforced: bool,
    /// `Transport.STAMP_LRS` default.
    pub(crate) stamp_link_requests: bool,
    /// `Transport.ALWAYS_STAMP_LRS` default.
    pub(crate) always_stamp_link_requests: bool,
    #[cfg(feature = "hooks")]
    pub(crate) provider_queue_max_events: usize,
    #[cfg(feature = "hooks")]
    pub(crate) provider_queue_max_bytes: usize,
}

#[cfg(feature = "iface-backbone")]
#[derive(Debug, Clone)]
pub(crate) struct BackboneDiscoveryRuntime {
    pub(crate) discoverable: bool,
    pub(crate) config: crate::discovery::DiscoveryConfig,
    pub(crate) transport_enabled: bool,
    pub(crate) ifac_netname: Option<String>,
    pub(crate) ifac_netkey: Option<String>,
}

#[cfg(feature = "iface-backbone")]
#[derive(Debug, Clone)]
pub(crate) struct BackboneDiscoveryRuntimeHandle {
    pub(crate) interface_name: String,
    pub(crate) current: BackboneDiscoveryRuntime,
    pub(crate) startup: BackboneDiscoveryRuntime,
}

#[cfg(feature = "iface-tcp")]
#[derive(Debug, Clone)]
pub(crate) struct TcpServerDiscoveryRuntime {
    pub(crate) discoverable: bool,
    pub(crate) config: crate::discovery::DiscoveryConfig,
    pub(crate) transport_enabled: bool,
    pub(crate) ifac_netname: Option<String>,
    pub(crate) ifac_netkey: Option<String>,
}

#[cfg(feature = "iface-tcp")]
#[derive(Debug, Clone)]
pub(crate) struct TcpServerDiscoveryRuntimeHandle {
    pub(crate) interface_name: String,
    pub(crate) current: TcpServerDiscoveryRuntime,
    pub(crate) startup: TcpServerDiscoveryRuntime,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct IfacRuntimeConfig {
    pub(crate) netname: Option<String>,
    pub(crate) netkey: Option<String>,
    pub(crate) size: usize,
}

#[cfg(feature = "iface-backbone")]
impl BackboneDiscoveryRuntimeHandle {
    pub(crate) fn from_parts(
        interface_name: String,
        startup_config: crate::discovery::DiscoveryConfig,
        transport_enabled: bool,
        ifac_netname: Option<String>,
        ifac_netkey: Option<String>,
        discoverable: bool,
    ) -> Self {
        let startup = BackboneDiscoveryRuntime {
            discoverable,
            config: startup_config,
            transport_enabled,
            ifac_netname,
            ifac_netkey,
        };
        Self {
            interface_name,
            current: startup.clone(),
            startup,
        }
    }
}

#[cfg(feature = "iface-tcp")]
impl TcpServerDiscoveryRuntimeHandle {
    pub(crate) fn from_parts(
        interface_name: String,
        startup_config: crate::discovery::DiscoveryConfig,
        transport_enabled: bool,
        ifac_netname: Option<String>,
        ifac_netkey: Option<String>,
        discoverable: bool,
    ) -> Self {
        let startup = TcpServerDiscoveryRuntime {
            discoverable,
            config: startup_config,
            transport_enabled,
            ifac_netname,
            ifac_netkey,
        };
        Self {
            interface_name,
            current: startup.clone(),
            startup,
        }
    }
}

impl IfacRuntimeConfig {
    pub(crate) fn from_parts(netname: Option<String>, netkey: Option<String>, size: usize) -> Self {
        Self {
            netname,
            netkey,
            size,
        }
    }
}

#[cfg(feature = "iface-backbone")]
#[derive(Debug, Clone)]
pub struct BackbonePeerPoolSettings {
    pub max_connected: usize,
    pub failure_threshold: usize,
    pub failure_window: Duration,
    pub cooldown: Duration,
}

#[cfg(feature = "iface-backbone")]
pub(crate) const BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY: u8 = 60;
#[cfg(feature = "iface-backbone")]
pub(crate) const BACKBONE_PEER_POOL_DISCOVERED_PRIORITY: u8 = 40;

#[cfg(feature = "iface-backbone")]
pub(crate) struct BackbonePeerPoolCandidateConfig {
    pub(crate) client: BackboneClientConfig,
    pub(crate) mode: u8,
    pub(crate) gravity: i64,
    pub(crate) recursive_prs: bool,
    pub(crate) announces_from_internal: bool,
    pub(crate) announces_to_internal: Option<bool>,
    pub(crate) ingress_control: rns_core::transport::types::IngressControlConfig,
    pub(crate) stamp_costs: rns_core::transport::types::InterfaceStampCosts,
    pub(crate) ifac_runtime: IfacRuntimeConfig,
    pub(crate) ifac_enabled: bool,
    pub(crate) interface_type_name: String,
    pub(crate) source: BackbonePeerPoolCandidateSource,
    pub(crate) priority: u8,
    pub(crate) discovery: Option<BackbonePeerPoolDiscoveryCandidate>,
}

#[cfg(feature = "iface-backbone")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BackbonePeerPoolCandidateSource {
    Configured,
    Discovered,
}

#[cfg(feature = "iface-backbone")]
impl BackbonePeerPoolCandidateSource {
    fn as_str(self) -> &'static str {
        match self {
            Self::Configured => "configured",
            Self::Discovered => "discovered",
        }
    }
}

#[cfg(feature = "iface-backbone")]
#[derive(Debug, Clone)]
pub(crate) struct BackbonePeerPoolDiscoveryCandidate {
    pub(crate) discovery_hash: [u8; 32],
    pub(crate) status: crate::discovery::DiscoveredStatus,
    pub(crate) hops: u8,
    pub(crate) stamp_value: rns_core::stamp::StampValue,
    pub(crate) last_heard: f64,
}

#[cfg(feature = "iface-backbone")]
struct BackbonePeerPool {
    settings: BackbonePeerPoolSettings,
    candidates: Vec<BackbonePeerPoolCandidate>,
}

#[cfg(feature = "iface-backbone")]
struct BackbonePeerPoolCandidate {
    config: BackbonePeerPoolCandidateConfig,
    active_id: Option<InterfaceId>,
    failures: Vec<f64>,
    /// Failed attempts since the last successful connect; drives the
    /// reconnect backoff of configured peers.
    consecutive_failures: u32,
    retry_after: Option<f64>,
    cooldown_until: Option<f64>,
    last_error: Option<String>,
    /// Stops the active connection's reader when the pool evicts it.
    control: Option<crate::interface::ListenerControl>,
}

/// Thin wrapper providing `EngineAccess` for a `TransportEngine` + Driver interfaces.
#[cfg(feature = "hooks")]
struct EngineRef<'a> {
    engine: &'a TransportEngine,
    interfaces: &'a crate::interface::InterfaceMap<InterfaceEntry>,
    link_manager: &'a LinkManager,
    now: f64,
}

#[cfg(feature = "hooks")]
impl<'a> EngineAccess for EngineRef<'a> {
    fn has_path(&self, dest: &[u8; 16]) -> bool {
        self.engine.has_path(dest)
    }
    fn hops_to(&self, dest: &[u8; 16]) -> Option<u8> {
        self.engine.hops_to(dest)
    }
    fn next_hop(&self, dest: &[u8; 16]) -> Option<[u8; 16]> {
        self.engine.next_hop(dest)
    }
    fn is_blackholed(&self, identity: &[u8; 16]) -> bool {
        self.engine.is_blackholed(identity, self.now)
    }
    fn interface_name(&self, id: u64) -> Option<String> {
        self.interfaces
            .get(&InterfaceId(id))
            .map(|e| e.info.name.clone())
    }
    fn interface_mode(&self, id: u64) -> Option<u8> {
        self.interfaces.get(&InterfaceId(id)).map(|e| e.info.mode)
    }
    fn identity_hash(&self) -> Option<[u8; 16]> {
        self.engine.identity_hash().copied()
    }
    fn announce_rate(&self, id: u64) -> Option<i32> {
        self.interfaces
            .get(&InterfaceId(id))
            .map(|e| (e.stats.outgoing_announce_freq() * 1000.0) as i32)
    }
    fn link_state(&self, link_hash: &[u8; 16]) -> Option<u8> {
        use rns_core::link::types::LinkState;
        self.link_manager.link_state(link_hash).map(|s| match s {
            LinkState::Pending => 0,
            LinkState::Handshake => 1,
            LinkState::Active => 2,
            LinkState::Stale => 3,
            LinkState::Closed => 4,
        })
    }
}

/// Extract the 16-byte destination hash from a raw packet header.
///
/// HEADER_1 (raw[0] & 0x40 == 0): dest at bytes 2..18
/// HEADER_2 (raw[0] & 0x40 != 0): dest at bytes 18..34 (after transport ID)
#[cfg(any(test, feature = "hooks"))]
fn extract_dest_hash(raw: &[u8]) -> [u8; 16] {
    let mut dest = [0u8; 16];
    if raw.is_empty() {
        return dest;
    }
    let is_header2 = raw[0] & 0x40 != 0;
    let start = if is_header2 { 18 } else { 2 };
    let end = start + 16;
    if raw.len() >= end {
        dest.copy_from_slice(&raw[start..end]);
    }
    dest
}

/// Execute a hook chain on disjoint Driver fields (avoids &mut self borrow conflict).
#[cfg(feature = "hooks")]
fn run_hook_inner(
    programs: &mut [rns_hooks::LoadedProgram],
    hook_manager: &Option<HookManager>,
    engine_access: &dyn EngineAccess,
    ctx: &HookContext,
    now: f64,
    provider_events_enabled: bool,
) -> Option<rns_hooks::ExecuteResult> {
    if programs.is_empty() {
        return None;
    }
    let mgr = hook_manager.as_ref()?;
    mgr.run_chain_with_provider_events(programs, ctx, engine_access, now, provider_events_enabled)
}

/// What a packet hook point does after its hook chain ran.
///
/// Only hook points that hold the raw packet bytes before acting on them apply
/// a `Modify` verdict (`PreIngress`, `SendOnInterface`,
/// `BroadcastOnAllInterfaces`, `DeliverLocal`). Every other hook point treats
/// `Modify` as `Continue` and ignores the replacement bytes.
#[cfg(feature = "hooks")]
pub(crate) enum PacketHookOutcome {
    /// Carry on with the bytes the hook point already holds.
    Keep,
    /// Stop processing this packet.
    Drop,
    /// Carry on with the replacement bytes, decoded with the same decoder the
    /// normal path uses. Hashes and header fields come from the replacement.
    Replace(RawPacket),
}

/// Validate replacement bytes from a `Modify` verdict: they must be non-empty,
/// fit within `max_len`, and decode as a packet.
#[cfg(feature = "hooks")]
fn validate_hook_replacement(data: &[u8], max_len: usize) -> Result<RawPacket, &'static str> {
    if data.is_empty() {
        return Err("replacement is empty");
    }
    if data.len() > max_len {
        return Err("replacement exceeds the packet size limit");
    }
    RawPacket::unpack(data).map_err(|_| "replacement does not decode as a packet")
}

#[cfg(feature = "hooks")]
fn backbone_peer_hook_context(event: &BackbonePeerHookEvent) -> HookContext<'_> {
    HookContext::BackbonePeer {
        server_interface_id: event.server_interface_id.0,
        peer_interface_id: event.peer_interface_id.map(|id| id.0),
        peer_ip: event.peer_ip,
        peer_port: event.peer_port,
        connected_for: event.connected_for,
        had_received_data: event.had_received_data,
        penalty_level: event.penalty_level,
        blacklist_for: event.blacklist_for,
    }
}

/// Convert a Vec of ActionWire into TransportActions for dispatch.
#[cfg(feature = "hooks")]
fn convert_injected_actions(actions: Vec<rns_hooks::ActionWire>) -> Vec<TransportAction> {
    actions
        .into_iter()
        .map(|a| {
            use rns_hooks::ActionWire;
            match a {
                ActionWire::SendOnInterface { interface, raw } => {
                    TransportAction::SendOnInterface {
                        interface: InterfaceId(interface),
                        raw: raw.into(),
                    }
                }
                ActionWire::BroadcastOnAllInterfaces {
                    raw,
                    exclude,
                    has_exclude,
                } => TransportAction::BroadcastOnAllInterfaces {
                    raw: raw.into(),
                    exclude: if has_exclude != 0 {
                        Some(InterfaceId(exclude))
                    } else {
                        None
                    },
                },
                ActionWire::DeliverLocal {
                    destination_hash,
                    raw,
                    packet_hash,
                    receiving_interface,
                } => TransportAction::DeliverLocal {
                    destination_hash,
                    raw: raw.into(),
                    packet_hash,
                    receiving_interface: InterfaceId(receiving_interface),
                },
                ActionWire::PathUpdated {
                    destination_hash,
                    hops,
                    next_hop,
                    interface,
                } => TransportAction::PathUpdated {
                    destination_hash,
                    hops,
                    next_hop,
                    interface: InterfaceId(interface),
                },
                ActionWire::CacheAnnounce { packet_hash, raw } => TransportAction::CacheAnnounce {
                    packet_hash,
                    raw: raw.into(),
                },
                ActionWire::TunnelEstablished {
                    tunnel_id,
                    interface,
                } => TransportAction::TunnelEstablished {
                    tunnel_id,
                    interface: InterfaceId(interface),
                },
                ActionWire::TunnelSynthesize {
                    interface,
                    data,
                    dest_hash,
                } => TransportAction::TunnelSynthesize {
                    interface: InterfaceId(interface),
                    data,
                    dest_hash,
                },
                ActionWire::ForwardToLocalClients {
                    raw,
                    exclude,
                    has_exclude,
                } => TransportAction::ForwardToLocalClients {
                    raw: raw.into(),
                    exclude: if has_exclude != 0 {
                        Some(InterfaceId(exclude))
                    } else {
                        None
                    },
                },
                ActionWire::ForwardPlainBroadcast {
                    raw,
                    to_local,
                    exclude,
                    has_exclude,
                } => TransportAction::ForwardPlainBroadcast {
                    raw: raw.into(),
                    to_local: to_local != 0,
                    exclude: if has_exclude != 0 {
                        Some(InterfaceId(exclude))
                    } else {
                        None
                    },
                },
                ActionWire::AnnounceReceived {
                    destination_hash,
                    identity_hash,
                    public_key,
                    name_hash,
                    random_hash,
                    app_data,
                    hops,
                    receiving_interface,
                } => TransportAction::AnnounceReceived(Box::new(ReceivedAnnounce {
                    destination_hash,
                    identity_hash,
                    public_key,
                    name_hash,
                    random_hash,
                    ratchet: None,
                    app_data,
                    stamp: None,
                    stamp_value: rns_core::stamp::StampValue::ZERO,
                    lr_cost: None,
                    hops,
                    receiving_interface: InterfaceId(receiving_interface),
                    rx: rns_core::transport::RxMetadata::default(),
                })),
            }
        })
        .collect()
}

/// Infer the interface type string from a dynamic interface's name.
/// Dynamic interfaces (TCP server clients, backbone peers, auto peers, local server clients)
/// include their type in the name prefix set at construction.
fn infer_interface_type(name: &str) -> String {
    if name.starts_with("TCPServerInterface") {
        "TCPServerClientInterface".to_string()
    } else if name.starts_with("BackboneInterface") {
        "BackboneInterface".to_string()
    } else if name.starts_with("LocalInterface") {
        "LocalServerClientInterface".to_string()
    } else if name.starts_with("WeaveInterfacePeer[") {
        "WeaveInterfacePeer".to_string()
    } else {
        // AutoInterface peers use "{group_name}:{peer_addr}" format where
        // group_name is the config section name (typically "AutoInterface" or similar).
        "AutoInterface".to_string()
    }
}

pub use crate::common::callbacks::Callbacks;

#[derive(Clone)]
struct SharedAnnounceRecord {
    name_hash: [u8; 10],
    identity_prv_key: [u8; 64],
    app_data: Option<Vec<u8>>,
}

/// Cache of path-response announces for local destinations:
/// (destination, path-request tag) → (created, raw announce).
type LocalPathResponseCache = HashMap<([u8; 16], Vec<u8>), (f64, Vec<u8>)>;

#[derive(Debug, Clone)]
pub(crate) struct KnownDestinationState {
    announced: crate::destination::AnnouncedIdentity,
    was_used: bool,
    last_used_at: Option<f64>,
    retained: bool,
    stamp: Option<[u8; rns_core::constants::STAMP_LENGTH]>,
    stamp_value: rns_core::stamp::StampValue,
    lr_cost: Option<rns_core::stamp::StampCost>,
}

struct KnownDestinationCleanup {
    candidates: Vec<[u8; 16]>,
    cursor: usize,
    started_at: Instant,
    removed: usize,
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct TrafficSample {
    sampled_at: f64,
    rxb: u64,
    txb: u64,
    rx_packets: u64,
    tx_packets: u64,
    arxb: u64,
    atxb: u64,
    prxb: u64,
    ptxb: u64,
}

impl TrafficSample {
    fn from_stats(sampled_at: f64, stats: &crate::interface::InterfaceStats) -> Self {
        Self {
            sampled_at,
            rxb: stats.rxb,
            txb: stats.txb,
            rx_packets: stats.rx_packets,
            tx_packets: stats.tx_packets,
            arxb: stats.arxb,
            atxb: stats.atxb,
            prxb: stats.prxb,
            ptxb: stats.ptxb,
        }
    }
}

/// The driver loop. Owns the engine and all interface entries.
pub(crate) struct ManagedInterface {
    pub(crate) parent_id: InterfaceId,
    pub(crate) type_name: String,
    pub(crate) control: Option<crate::interface::ListenerControl>,
    pub(crate) static_ids: Vec<InterfaceId>,
}

/// Interval of the transport-state snapshot, as the persistence thread keeps.
pub(crate) const TRANSPORT_STATE_PERSIST_INTERVAL: Duration = Duration::from_secs(12 * 60 * 60);

/// Deadlines and pending jobs of the timer and housekeeping work the driver
/// runs itself when [`crate::engine::FOLD_TIMERS`] is set.
pub(crate) struct FoldedTimers {
    pub(crate) next_tick: Instant,
    pub(crate) next_persist: Instant,
    /// Where transport state is persisted, on transport nodes.
    pub(crate) persist_dir: Option<std::path::PathBuf>,
    pub(crate) jobs: std::collections::VecDeque<FoldedJob>,
    /// Whether a timerfd loop source delivers the tick.
    pub(crate) ticks_from_source: bool,
    /// Announces taken from the verification queue, checked one per idle
    /// turn when CPU work is folded.
    pub(crate) announces: std::collections::VecDeque<(
        rns_core::transport::announce_verify_queue::AnnounceVerifyKey,
        rns_core::transport::announce_verify_queue::PendingAnnounce,
    )>,
    /// Whether the verification queue may hold announces not yet taken.
    pub(crate) announces_queued: bool,
    /// Outgoing resources compressed on the driver loop, oldest first.
    pub(crate) compressions: std::collections::VecDeque<resource_preparation::FoldedCompression>,
    pub(crate) next_compression_id: u64,
}

pub(crate) enum FoldedJob {
    PersistTransportState,
    CleanRatchets {
        known_destinations: HashSet<[u8; 16]>,
        now: f64,
        expiry: f64,
    },
}

impl FoldedTimers {
    fn new() -> Self {
        let now = Instant::now();
        FoldedTimers {
            next_tick: now + Duration::from_millis(DEFAULT_TICK_INTERVAL_MS),
            next_persist: now + TRANSPORT_STATE_PERSIST_INTERVAL,
            persist_dir: None,
            jobs: std::collections::VecDeque::new(),
            ticks_from_source: false,
            announces: std::collections::VecDeque::new(),
            announces_queued: false,
            compressions: std::collections::VecDeque::new(),
            next_compression_id: 0,
        }
    }
}

pub struct Driver {
    resource_preparation: Option<resource_preparation::Worker>,
    pub(crate) tracked_link_send: Option<([u8; 32], crate::link_send::Completion)>,
    pub(crate) pending_link_frames: std::collections::VecDeque<PendingLinkFrame>,
    pub(crate) engine: TransportEngine,
    pub(crate) interfaces: crate::interface::InterfaceMap<InterfaceEntry>,
    /// Parent listener for dynamically spawned interfaces.
    pub(crate) dynamic_interface_parents: HashMap<InterfaceId, InterfaceId>,
    /// Previous cumulative counters used to calculate current one-second rates.
    pub(crate) traffic_samples: HashMap<InterfaceId, TrafficSample>,
    pub(crate) rng: OsRng,
    pub(crate) rx: EventReceiver,
    pub(crate) callbacks: Box<dyn Callbacks>,
    pub(crate) started: f64,
    pub(crate) lifecycle_state: LifecycleState,
    /// Whether `lifecycle_state` is still `Active`, shared with the node
    /// handle so that each API call can check it without a driver query.
    pub(crate) accepting_work: Arc<AtomicBool>,
    /// This node is a client of a shared instance, which sends the path
    /// requests that rediscover paths for it
    /// (`Transport.owner.is_connected_to_shared_instance`).
    pub(crate) connected_to_shared_instance: bool,
    pub(crate) drain_started_at: Option<Instant>,
    pub(crate) drain_deadline: Option<Instant>,
    pub(crate) listener_controls: Vec<crate::interface::ListenerControl>,
    pub(crate) interface_management: Option<crate::node::InterfaceManagementConfig>,
    pub(crate) managed_interfaces: HashMap<String, ManagedInterface>,
    pub(crate) retired_interface_parents: HashSet<InterfaceId>,
    /// Writers that may hold frames queued without a wakeup.
    pub(crate) writers_to_wake: Vec<InterfaceId>,
    /// Ingress burst state last published to the event queue, per interface.
    pub(crate) published_ingress_bursts: crate::interface::InterfaceMap<(Option<f64>, Option<f64>)>,
    pub(crate) announce_cache: Option<crate::announce_cache::AnnounceCache>,
    /// Destination hash for rnstransport.tunnel.synthesize (PLAIN).
    pub(crate) tunnel_synth_dest: [u8; 16],
    /// Transport identity (optional, needed for tunnel synthesis).
    pub(crate) transport_identity: Option<rns_crypto::identity::Identity>,
    /// Link manager: handles link lifecycle, request/response.
    pub(crate) link_manager: LinkManager,
    /// Management configuration for ACL checks.
    pub(crate) management_config: crate::management::ManagementConfig,
    /// Last time management announces were emitted.
    pub(crate) last_management_announce: f64,
    /// Whether initial management announce has been sent (delayed 5s after start).
    pub(crate) initial_announce_sent: bool,
    /// Cache of known announced identities and lifecycle state, keyed by destination hash.
    pub(crate) known_destinations: HashMap<[u8; 16], KnownDestinationState>,
    /// Store for received remote ratchets, if persistence/use is enabled.
    pub(crate) ratchet_store: Option<Arc<dyn crate::storage::RatchetStore>>,
    /// TTL for received ratchets, in seconds.
    pub(crate) ratchet_expiry_secs: f64,
    /// TTL for known destinations without an active path, in seconds.
    pub(crate) known_destinations_ttl: f64,
    /// Maximum number of retained known destinations.
    pub(crate) known_destinations_max_entries: usize,
    /// TTL for announce rate-limiter entries without an active path, in seconds.
    pub(crate) rate_limiter_ttl_secs: f64,
    /// Destination hash for rnstransport.path.request (PLAIN).
    pub(crate) path_request_dest: [u8; 16],
    /// Proof strategies per destination hash.
    /// Maps dest_hash → (strategy, optional signing identity for generating proofs).
    pub(crate) proof_strategies: HashMap<
        [u8; 16],
        (
            rns_core::types::ProofStrategy,
            Option<rns_crypto::identity::Identity>,
        ),
    >,
    /// Tracked sent packets for proof matching: packet_hash → (dest_hash, sent_time).
    pub(crate) sent_packets: HashMap<[u8; 32], ([u8; 16], f64)>,
    /// Local destinations whose single packets are delivered decrypted, with
    /// the identity they are decrypted with when they have no ratchets.
    pub(crate) decrypted_destinations: HashMap<[u8; 16], Arc<rns_crypto::identity::Identity>>,
    /// Reused buffer for single-packet proofs the proof pool finished.
    pub(crate) packet_proofs: Vec<crate::common::link_manager::ProofOutcome>,
    /// Reused transport action buffers (see [`crate::action_pool`]).
    pub(crate) transport_actions: crate::action_pool::ActionPool<TransportAction>,
    /// Reused link manager action buffers.
    pub(crate) link_actions: crate::action_pool::ActionPool<LinkManagerAction>,
    /// Completed proofs for probe polling: packet_hash → (rtt_seconds, received_time).
    /// Only probes ([`QueryRequest::SendProbe`]) are recorded: a proof for
    /// every other packet would be kept two minutes for nobody to poll.
    pub(crate) completed_proofs: HashMap<[u8; 32], (f64, f64)>,
    /// Probes sent and not yet proved, whose proofs `completed_proofs`
    /// records.
    pub(crate) pending_probes: std::collections::HashSet<[u8; 32]>,
    /// Parsed signing keys of destinations whose packet proofs were checked.
    /// A destination's hash follows from its identity, so its key never
    /// changes.
    pub(crate) proof_keys: HashMap<[u8; 16], rns_crypto::ed25519::Ed25519PublicKey>,
    /// Locally registered destinations: hash → dest_type.
    pub(crate) local_destinations: HashMap<[u8; 16], u8>,
    /// Latest explicit SINGLE announces to replay after shared-client reconnect.
    shared_announces: HashMap<[u8; 16], SharedAnnounceRecord>,
    /// Shared local interfaces that went down and should replay announces on reconnect.
    shared_reconnect_pending: HashMap<InterfaceId, bool>,
    /// Path-response announces recently sent for local destinations, keyed by
    /// (destination, path-request tag), replayed byte-for-byte for a repeated
    /// tag within `LOCAL_PATH_RESPONSE_WINDOW`: (created, raw announce).
    pub(crate) local_path_responses: LocalPathResponseCache,
    /// Hole-punch manager for direct P2P connections.
    pub(crate) holepunch_manager: HolePunchManager,
    /// Event sender for worker threads to send results back to the driver loop.
    pub(crate) event_tx: crate::event::EventSender,
    /// Grinds own path request and announce stamps off the driver thread.
    /// Replaced by the node's shared instance at startup.
    pub(crate) stamper: Arc<crate::stamper::Stamper>,
    /// `Transport.STAMP_DESTINATIONS`, shared with the node handle whose
    /// announce API reads it; settable as `global.stamp_destinations`.
    pub(crate) stamp_destinations: Arc<AtomicBool>,
    /// The `[stamps]` config section this node started with. Link request
    /// stamping reads `stamp_lrs`, `always_stamp_lrs` and `lr_stamp_timeout`
    /// from it; the other
    /// options are applied to the engine, the stamper and the node handle at
    /// startup.
    pub(crate) stamp_settings: crate::config::StampsSection,
    /// Inbound packets whose hop count reaches this limit are dropped and
    /// counted as `hop_limiter_hits` (`Transport.HOP_LIMITER`, set from
    /// `[reticulum] hop_limiter`).
    pub(crate) hop_limiter: u8,
    /// Maximum queued outbound frames per interface writer worker.
    pub(crate) interface_writer_queue_capacity: usize,
    /// Linux mark inherited by dynamically created IP underlay sockets.
    pub(crate) underlay_mark: Option<u32>,
    /// Default announce-rate controls for interfaces on transport nodes.
    pub(crate) announce_rate_defaults: AnnounceRateDefaults,
    /// Default ingress/egress controls for interfaces on transport nodes.
    pub(crate) ingress_control_defaults: rns_core::transport::types::IngressControlConfig,
    /// Shared timer interval used by the node timer thread.
    pub(crate) tick_interval_ms: Arc<AtomicU64>,
    /// Runtime-config handles for backbone server interfaces, keyed by config name.
    #[cfg(feature = "iface-backbone")]
    pub(crate) backbone_runtime: HashMap<String, BackboneRuntimeConfigHandle>,
    /// Live peer-state handles for backbone server interfaces, keyed by config name.
    #[cfg(feature = "iface-backbone")]
    pub(crate) backbone_peer_state: HashMap<String, BackbonePeerStateHandle>,
    /// Runtime-config handles for backbone client interfaces, keyed by config name.
    #[cfg(feature = "iface-backbone")]
    pub(crate) backbone_client_runtime: HashMap<String, BackboneClientRuntimeConfigHandle>,
    /// Runtime-config state for backbone discovery metadata, keyed by config name.
    #[cfg(feature = "iface-backbone")]
    pub(crate) backbone_discovery_runtime: HashMap<String, BackboneDiscoveryRuntimeHandle>,
    /// Ordered outbound Backbone peer pool, if enabled.
    #[cfg(feature = "iface-backbone")]
    backbone_peer_pool: Option<BackbonePeerPool>,
    /// Shared allocator for listener-created and discovery-created dynamic interfaces.
    pub(crate) next_dynamic_interface_id: Arc<AtomicU64>,
    /// Runtime-config handles for TCP server interfaces, keyed by config name.
    #[cfg(feature = "iface-tcp")]
    pub(crate) tcp_server_runtime: HashMap<String, TcpServerRuntimeConfigHandle>,
    /// Runtime-config handles for TCP client interfaces, keyed by config name.
    #[cfg(feature = "iface-tcp")]
    pub(crate) tcp_client_runtime: HashMap<String, TcpClientRuntimeConfigHandle>,
    /// Runtime-config state for TCP server discovery metadata, keyed by config name.
    #[cfg(feature = "iface-tcp")]
    pub(crate) tcp_server_discovery_runtime: HashMap<String, TcpServerDiscoveryRuntimeHandle>,
    /// Runtime-config handles for UDP interfaces, keyed by config name.
    #[cfg(feature = "iface-udp")]
    pub(crate) udp_runtime: HashMap<String, UdpRuntimeConfigHandle>,
    /// Runtime-config handles for Auto interfaces, keyed by config name.
    #[cfg(feature = "iface-auto")]
    pub(crate) auto_runtime: HashMap<String, AutoRuntimeConfigHandle>,
    /// Runtime-config handles for I2P interfaces, keyed by config name.
    #[cfg(feature = "iface-i2p")]
    pub(crate) i2p_runtime: HashMap<String, I2pRuntimeConfigHandle>,
    /// Runtime-config handles for Pipe interfaces, keyed by config name.
    #[cfg(feature = "iface-pipe")]
    pub(crate) pipe_runtime: HashMap<String, PipeRuntimeConfigHandle>,
    /// Runtime-config handles for RNode interfaces, keyed by config name.
    #[cfg(feature = "iface-rnode")]
    pub(crate) rnode_runtime: HashMap<String, RNodeRuntimeConfigHandle>,
    /// Startup/default interface metadata for generic cross-cutting runtime config.
    pub(crate) interface_runtime_defaults:
        HashMap<String, rns_core::transport::types::InterfaceInfo>,
    /// Current IFAC runtime config for static interfaces that support IFAC mutation.
    pub(crate) interface_ifac_runtime: HashMap<String, IfacRuntimeConfig>,
    /// Startup/default IFAC runtime config for static interfaces.
    pub(crate) interface_ifac_runtime_defaults: HashMap<String, IfacRuntimeConfig>,
    /// Storage for discovered interfaces.
    pub(crate) discovered_interfaces: crate::discovery::DiscoveredInterfaceStorage,
    /// Required stamp value for accepting discovered interfaces.
    pub(crate) discovery_required_value: rns_core::stamp::StampCost,
    /// Bounded cache of discovery payloads with previously rejected stamps.
    pub(crate) discovery_stamp_cache: crate::discovery::DiscoveryStampCache,
    /// Name hash for interface discovery announces ("rnstransport.discovery.interface").
    pub(crate) discovery_name_hash: [u8; 10],
    /// Destination hash for the probe responder (if respond_to_probes is enabled).
    pub(crate) probe_responder_hash: Option<[u8; 16]>,
    /// Whether interface discovery is enabled.
    pub(crate) discover_interfaces: bool,
    /// Explicit mode for interfaces created from discovery announces.
    pub(crate) autoconnect_interface_mode: Option<u8>,
    /// Gravity for interfaces created from discovery announces.
    pub(crate) autoconnect_interface_gravity: i64,
    /// Internal announce propagation override for discovered interfaces.
    pub(crate) autoconnect_announces_to_internal: Option<bool>,
    /// Allow auto-connecting discovered interfaces from unverified implementations.
    pub(crate) autoconnect_unverified_implementations: bool,
    /// Announcer for discoverable interfaces (None if nothing to announce).
    pub(crate) interface_announcer: Option<crate::discovery::InterfaceAnnouncer>,
    /// Shared async announce verification queue.
    pub(crate) announce_verify_queue: Arc<Mutex<AnnounceVerifyQueue>>,
    /// Wakes the verification thread when announces are queued.
    pub(crate) announce_verify_signal: Arc<WorkSignal>,
    /// Whether inbound announces should be verified off the driver thread.
    pub(crate) async_announce_verification: bool,
    /// Tick counter for periodic discovery cleanup (every ~3600 ticks = ~1 hour).
    pub(crate) discovery_cleanup_counter: u32,
    /// Runtime-configurable discovery cleanup interval.
    pub(crate) discovery_cleanup_interval_ticks: u32,
    /// Tick counter for periodic MEMSTATS logging (every 300 ticks = ~5 min).
    pub(crate) memory_stats_counter: u32,
    /// When to republish the link/interface statistics snapshot.
    pub(crate) state_snapshot_schedule: crate::state_snapshot::PublishSchedule,
    /// Tick counter for periodic memory/cache cleanup (every ~3600 ticks = ~1 hour).
    pub(crate) cache_cleanup_counter: u32,
    /// Tick counter for incremental announce-cache cleanup scheduling.
    pub(crate) announce_cache_cleanup_counter: u32,
    /// Runtime-configurable cleanup interval for known destinations.
    pub(crate) known_destinations_cleanup_interval_ticks: u32,
    /// Incremental known-destination cleanup pass, if one is active.
    known_destination_cleanup: Option<KnownDestinationCleanup>,
    /// Low-priority ratchet filesystem cleanup associated with the last pass.
    ratchet_cleanup_handle: Option<std::thread::JoinHandle<()>>,
    /// Timers and housekeeping the driver runs itself when they are folded
    /// (see [`crate::engine::FOLD_TIMERS`]).
    pub(crate) folded: FoldedTimers,
    /// The RPC servers, when the driver serves them itself (see
    /// [`crate::engine::FOLD_SERVICES`]).
    pub(crate) rpc_service: Option<rpc_service::RpcService>,
    /// The probe server, when it runs on the driver loop (with services
    /// folded); otherwise the node runs it on a thread of its own.
    pub(crate) probe_server: Option<crate::holepunch::probe::FoldedProbeServer>,
    /// Count of known-destination cap evictions since start.
    pub(crate) known_destinations_cap_evict_count: usize,
    /// Runtime-configurable interval for starting announce cache cleanup.
    pub(crate) announce_cache_cleanup_interval_ticks: u32,
    /// When set, announce cache cleanup is in progress (contains active packet hashes).
    /// Directory iterator for incremental announce cache cleanup.
    /// Running total of files removed during current cache cleanup cycle.
    /// Runtime-configurable announce cache cleanup batch size.
    pub(crate) announce_cache_cleanup_batch_size: usize,
    /// Runtime-configurable management announce interval.
    pub(crate) management_announce_interval_secs: f64,
    /// Startup/default runtime-config values.
    pub(crate) runtime_config_defaults: RuntimeConfigDefaults,
    /// Hook slots for the programmable hook system (one per HookPoint).
    #[cfg(feature = "hooks")]
    pub(crate) hook_slots: [HookSlot; HookPoint::COUNT],
    /// Hook manager. None if initialization failed.
    #[cfg(feature = "hooks")]
    pub(crate) hook_manager: Option<HookManager>,
    #[cfg(feature = "hooks")]
    pub(crate) provider_bridge: Option<ProviderBridge>,
    /// Per hook point count of `Modify` replacements that failed validation
    /// and were therefore treated as `Drop`.
    #[cfg(feature = "hooks")]
    pub(crate) hook_rejected_modifications: [u64; HookPoint::COUNT],
}

impl Driver {
    /// Create a new driver.
    pub fn new(
        config: TransportConfig,
        rx: EventReceiver,
        tx: crate::event::EventSender,
        callbacks: Box<dyn Callbacks>,
    ) -> Self {
        let announce_queue_max_entries = config.announce_queue_max_entries;
        let tunnel_synth_dest = rns_core::destination::destination_hash(
            "rnstransport",
            &["tunnel", "synthesize"],
            None,
        );
        let path_request_dest =
            rns_core::destination::destination_hash("rnstransport", &["path", "request"], None);
        let discovery_name_hash = crate::discovery::discovery_name_hash();
        let mut engine = TransportEngine::new(config);
        engine.seed_table_hashing(&mut OsRng);
        engine.register_destination(tunnel_synth_dest, rns_core::constants::DESTINATION_PLAIN);
        // Register path request destination so inbound path requests are delivered locally
        engine.register_destination(path_request_dest, rns_core::constants::DESTINATION_PLAIN);
        // Note: discovery destination is NOT registered as local — it's a SINGLE destination
        // whose hash depends on the sender's identity. We match it by name_hash instead.
        let mut local_destinations = HashMap::new();
        local_destinations.insert(tunnel_synth_dest, rns_core::constants::DESTINATION_PLAIN);
        local_destinations.insert(path_request_dest, rns_core::constants::DESTINATION_PLAIN);
        let runtime_config_defaults = RuntimeConfigDefaults {
            tick_interval_ms: DEFAULT_TICK_INTERVAL_MS,
            known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
            rate_limiter_ttl_secs: DEFAULT_RATE_LIMITER_TTL_SECS,
            known_destinations_cleanup_interval_ticks:
                DEFAULT_KNOWN_DESTINATIONS_CLEANUP_INTERVAL_TICKS,
            announce_cache_cleanup_interval_ticks: DEFAULT_ANNOUNCE_CACHE_CLEANUP_INTERVAL_TICKS,
            announce_cache_cleanup_batch_size: DEFAULT_ANNOUNCE_CACHE_CLEANUP_BATCH_SIZE,
            discovery_cleanup_interval_ticks: DEFAULT_DISCOVERY_CLEANUP_INTERVAL_TICKS,
            management_announce_interval_secs: DEFAULT_MANAGEMENT_ANNOUNCE_INTERVAL_SECS,
            direct_connect_policy: crate::event::HolePunchPolicy::default(),
            stamp_destinations: rns_core::constants::STAMP_DESTINATIONS,
            stamp_path_requests: rns_core::constants::STAMP_PRS,
            stamps_enforced: rns_core::constants::STAMPS_ENFORCED,
            lr_stamps_enforced: rns_core::constants::LR_STAMPS_ENFORCED,
            stamp_link_requests: rns_core::constants::STAMP_LRS,
            always_stamp_link_requests: rns_core::constants::ALWAYS_STAMP_LRS,
            #[cfg(feature = "hooks")]
            provider_queue_max_events: crate::provider_bridge::ProviderBridgeConfig::default()
                .queue_max_events,
            #[cfg(feature = "hooks")]
            provider_queue_max_bytes: crate::provider_bridge::ProviderBridgeConfig::default()
                .queue_max_bytes,
        };
        Driver {
            tracked_link_send: None,
            pending_link_frames: std::collections::VecDeque::new(),
            engine,
            interfaces: Default::default(),
            dynamic_interface_parents: HashMap::new(),
            traffic_samples: HashMap::new(),
            rng: OsRng,
            rx,
            callbacks,
            started: time::now(),
            lifecycle_state: LifecycleState::Active,
            accepting_work: Arc::new(AtomicBool::new(true)),
            connected_to_shared_instance: false,
            drain_started_at: None,
            drain_deadline: None,
            listener_controls: Vec::new(),
            interface_management: None,
            managed_interfaces: HashMap::new(),
            retired_interface_parents: HashSet::new(),
            published_ingress_bursts: Default::default(),
            writers_to_wake: Vec::new(),
            announce_cache: None,
            tunnel_synth_dest,
            transport_identity: None,
            link_manager: {
                let mut link_manager = LinkManager::new();
                link_manager.set_link_request_costs(tx.link_request_costs().clone());
                link_manager
            },
            management_config: Default::default(),
            last_management_announce: 0.0,
            initial_announce_sent: false,
            known_destinations: HashMap::new(),
            ratchet_store: None,
            ratchet_expiry_secs: rns_core::constants::RATCHET_EXPIRY as f64,
            known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
            known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
            rate_limiter_ttl_secs: DEFAULT_RATE_LIMITER_TTL_SECS,
            path_request_dest,
            proof_strategies: HashMap::new(),
            sent_packets: HashMap::new(),
            packet_proofs: Vec::new(),
            decrypted_destinations: HashMap::new(),
            transport_actions: crate::action_pool::ActionPool::new(),
            link_actions: crate::action_pool::ActionPool::new(),
            completed_proofs: HashMap::new(),
            pending_probes: std::collections::HashSet::new(),
            proof_keys: HashMap::new(),
            local_destinations,
            shared_announces: HashMap::new(),
            shared_reconnect_pending: HashMap::new(),
            local_path_responses: HashMap::new(),
            holepunch_manager: {
                let mut manager =
                    HolePunchManager::new(vec![], rns_core::holepunch::ProbeProtocol::Rnsp, None);
                manager.set_event_sender(tx.clone());
                manager
            },
            stamper: Arc::new(crate::stamper::Stamper::new(tx.clone(), None)),
            stamp_destinations: Arc::new(AtomicBool::new(rns_core::constants::STAMP_DESTINATIONS)),
            stamp_settings: crate::config::StampsSection::default(),
            hop_limiter: rns_core::constants::HOP_LIMITER,
            event_tx: tx,
            interface_writer_queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
            underlay_mark: None,
            announce_rate_defaults: AnnounceRateDefaults::default(),
            ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(),
            tick_interval_ms: Arc::new(AtomicU64::new(DEFAULT_TICK_INTERVAL_MS)),
            #[cfg(feature = "iface-backbone")]
            backbone_runtime: HashMap::new(),
            #[cfg(feature = "iface-backbone")]
            backbone_peer_state: HashMap::new(),
            #[cfg(feature = "iface-backbone")]
            backbone_client_runtime: HashMap::new(),
            #[cfg(feature = "iface-backbone")]
            backbone_discovery_runtime: HashMap::new(),
            #[cfg(feature = "iface-backbone")]
            backbone_peer_pool: None,
            next_dynamic_interface_id: Arc::new(AtomicU64::new(10000)),
            #[cfg(feature = "iface-tcp")]
            tcp_server_runtime: HashMap::new(),
            #[cfg(feature = "iface-tcp")]
            tcp_client_runtime: HashMap::new(),
            #[cfg(feature = "iface-tcp")]
            tcp_server_discovery_runtime: HashMap::new(),
            #[cfg(feature = "iface-udp")]
            udp_runtime: HashMap::new(),
            #[cfg(feature = "iface-auto")]
            auto_runtime: HashMap::new(),
            #[cfg(feature = "iface-i2p")]
            i2p_runtime: HashMap::new(),
            #[cfg(feature = "iface-pipe")]
            pipe_runtime: HashMap::new(),
            #[cfg(feature = "iface-rnode")]
            rnode_runtime: HashMap::new(),
            interface_runtime_defaults: HashMap::new(),
            interface_ifac_runtime: HashMap::new(),
            interface_ifac_runtime_defaults: HashMap::new(),
            discovered_interfaces: crate::discovery::DiscoveredInterfaceStorage::new(
                std::env::temp_dir().join("rns-discovered-interfaces"),
            ),
            discovery_required_value: crate::discovery::DEFAULT_STAMP_VALUE,
            discovery_stamp_cache: crate::discovery::DiscoveryStampCache::default(),
            discovery_name_hash,
            probe_responder_hash: None,
            discover_interfaces: false,
            autoconnect_interface_mode: None,
            autoconnect_interface_gravity: 0,
            autoconnect_announces_to_internal: None,
            autoconnect_unverified_implementations: false,
            interface_announcer: None,
            announce_verify_queue: Arc::new(Mutex::new(AnnounceVerifyQueue::new(
                announce_queue_max_entries,
            ))),
            announce_verify_signal: Arc::default(),
            async_announce_verification: false,
            discovery_cleanup_counter: 0,
            discovery_cleanup_interval_ticks: runtime_config_defaults
                .discovery_cleanup_interval_ticks,
            memory_stats_counter: 0,
            state_snapshot_schedule: Default::default(),
            cache_cleanup_counter: 0,
            announce_cache_cleanup_counter: 0,
            known_destinations_cleanup_interval_ticks: runtime_config_defaults
                .known_destinations_cleanup_interval_ticks,
            known_destination_cleanup: None,
            ratchet_cleanup_handle: None,
            folded: FoldedTimers::new(),
            rpc_service: None,
            probe_server: None,
            resource_preparation: None,
            known_destinations_cap_evict_count: 0,
            announce_cache_cleanup_interval_ticks: runtime_config_defaults
                .announce_cache_cleanup_interval_ticks,
            announce_cache_cleanup_batch_size: runtime_config_defaults
                .announce_cache_cleanup_batch_size,
            management_announce_interval_secs: runtime_config_defaults
                .management_announce_interval_secs,
            runtime_config_defaults,
            #[cfg(feature = "hooks")]
            hook_slots: create_hook_slots(),
            #[cfg(feature = "hooks")]
            hook_manager: HookManager::new().ok(),
            #[cfg(feature = "hooks")]
            provider_bridge: None,
            #[cfg(feature = "hooks")]
            hook_rejected_modifications: [0; HookPoint::COUNT],
        }
    }

    pub fn set_announce_verify_queue_config(
        &mut self,
        max_entries: usize,
        max_bytes: usize,
        max_stale_secs: f64,
        overflow_policy: OverflowPolicy,
    ) {
        self.announce_verify_queue = Arc::new(Mutex::new(AnnounceVerifyQueue::with_limits(
            max_entries,
            max_bytes,
            max_stale_secs,
            overflow_policy,
        )));
    }

    pub fn set_announce_rate_defaults(&mut self, defaults: AnnounceRateDefaults) {
        self.announce_rate_defaults = defaults;
    }

    pub fn set_ingress_control_defaults(
        &mut self,
        defaults: rns_core::transport::types::IngressControlConfig,
    ) {
        self.ingress_control_defaults = defaults;
    }

    pub(crate) fn apply_announce_rate_defaults(
        &self,
        info: &mut rns_core::transport::types::InterfaceInfo,
    ) {
        if !self.engine.transport_enabled() || info.announce_rate_target.is_some() {
            return;
        }

        let Some(target) = self.announce_rate_defaults.target else {
            return;
        };

        info.announce_rate_target = Some(target);
        if info.announce_rate_grace == 0 {
            info.announce_rate_grace = self.announce_rate_defaults.grace;
        }
        if info.announce_rate_penalty == 0.0 {
            info.announce_rate_penalty = self.announce_rate_defaults.penalty;
        }
    }

    pub(crate) fn apply_ingress_control_defaults(
        &self,
        info: &mut rns_core::transport::types::InterfaceInfo,
    ) {
        if !self.engine.transport_enabled() {
            return;
        }

        let baseline = if info.ingress_control.enabled {
            rns_core::transport::types::IngressControlConfig::enabled()
        } else {
            rns_core::transport::types::IngressControlConfig::disabled()
        };
        let defaults = self.ingress_control_defaults;
        let ic = &mut info.ingress_control;

        if ic.max_held_announces == baseline.max_held_announces {
            ic.max_held_announces = defaults.max_held_announces;
        }
        if ic.burst_hold == baseline.burst_hold {
            ic.burst_hold = defaults.burst_hold;
        }
        if ic.burst_freq_new == baseline.burst_freq_new {
            ic.burst_freq_new = defaults.burst_freq_new;
        }
        if ic.burst_freq == baseline.burst_freq {
            ic.burst_freq = defaults.burst_freq;
        }
        if ic.pr_burst_freq_new == baseline.pr_burst_freq_new {
            ic.pr_burst_freq_new = defaults.pr_burst_freq_new;
        }
        if ic.pr_burst_freq == baseline.pr_burst_freq {
            ic.pr_burst_freq = defaults.pr_burst_freq;
        }
        if ic.new_time == baseline.new_time {
            ic.new_time = defaults.new_time;
        }
        if ic.burst_penalty == baseline.burst_penalty {
            ic.burst_penalty = defaults.burst_penalty;
        }
        if ic.held_release_interval == baseline.held_release_interval {
            ic.held_release_interval = defaults.held_release_interval;
        }
        if ic.egress_enabled == baseline.egress_enabled {
            ic.egress_enabled = defaults.egress_enabled;
        }
        if ic.egress_pr_freq == baseline.egress_pr_freq {
            ic.egress_pr_freq = defaults.egress_pr_freq;
        }
    }

    fn wrap_interface_writer(
        &self,
        interface_id: InterfaceId,
        interface_name: &str,
        writer: Box<dyn crate::interface::Writer>,
    ) -> (
        Box<dyn crate::interface::Writer>,
        crate::interface::AsyncWriterMetrics,
    ) {
        crate::interface::wrap_async_writer(
            writer,
            interface_id,
            interface_name,
            self.event_tx.clone(),
            self.interface_writer_queue_capacity,
        )
    }

    #[cfg(feature = "hooks")]
    fn provider_events_enabled(&self) -> bool {
        self.provider_bridge.is_some()
    }

    /// Whether any program is attached at `point`. Per-packet hook points
    /// check this before building a hook context, which reads the clock.
    #[cfg(feature = "hooks")]
    pub(crate) fn hooks_attached(&self, point: HookPoint) -> bool {
        !self.hook_slots[point as usize].programs.is_empty()
    }

    #[cfg(feature = "hooks")]
    fn run_backbone_peer_hook(
        &mut self,
        attach_point: &str,
        point: HookPoint,
        event: &BackbonePeerHookEvent,
    ) {
        let ctx = backbone_peer_hook_context(event);
        let now = time::now();
        let engine_ref = EngineRef {
            engine: &self.engine,
            interfaces: &self.interfaces,
            link_manager: &self.link_manager,
            now,
        };
        let provider_events_enabled = self.provider_events_enabled();
        if let Some(ref e) = run_hook_inner(
            &mut self.hook_slots[point as usize].programs,
            &self.hook_manager,
            &engine_ref,
            &ctx,
            now,
            provider_events_enabled,
        ) {
            self.forward_hook_side_effects(attach_point, e);
        }
    }

    #[cfg(feature = "iface-backbone")]
    fn make_discoverable_interface(
        runtime: &BackboneDiscoveryRuntimeHandle,
    ) -> crate::discovery::DiscoverableInterface {
        crate::discovery::DiscoverableInterface {
            interface_name: runtime.interface_name.clone(),
            config: runtime.current.config.clone(),
            transport_enabled: runtime.current.transport_enabled,
            ifac_netname: runtime.current.ifac_netname.clone(),
            ifac_netkey: runtime.current.ifac_netkey.clone(),
        }
    }

    #[cfg(feature = "iface-backbone")]
    fn sync_backbone_discovery_runtime(
        &mut self,
        interface_name: &str,
    ) -> Result<(), RuntimeConfigError> {
        let handle = self
            .backbone_discovery_runtime
            .get(interface_name)
            .ok_or(RuntimeConfigError {
                code: RuntimeConfigErrorCode::NotFound,
                message: format!("backbone interface '{}' not found", interface_name),
            })?
            .clone();

        if handle.current.discoverable {
            let iface = Self::make_discoverable_interface(&handle);
            if let Some(announcer) = self.interface_announcer.as_mut() {
                announcer.upsert_interface(iface);
            } else if let Some(identity) = self.transport_identity.as_ref() {
                self.interface_announcer = Some(crate::discovery::InterfaceAnnouncer::new(
                    *identity.hash(),
                    vec![iface],
                ));
            }
        } else if let Some(announcer) = self.interface_announcer.as_mut() {
            announcer.remove_interface(interface_name);
            if announcer.is_empty() {
                self.interface_announcer = None;
            }
        }

        Ok(())
    }

    #[cfg(feature = "iface-tcp")]
    fn make_tcp_server_discoverable_interface(
        runtime: &TcpServerDiscoveryRuntimeHandle,
    ) -> crate::discovery::DiscoverableInterface {
        crate::discovery::DiscoverableInterface {
            interface_name: runtime.interface_name.clone(),
            config: runtime.current.config.clone(),
            transport_enabled: runtime.current.transport_enabled,
            ifac_netname: runtime.current.ifac_netname.clone(),
            ifac_netkey: runtime.current.ifac_netkey.clone(),
        }
    }

    #[cfg(feature = "iface-tcp")]
    fn sync_tcp_server_discovery_runtime(
        &mut self,
        interface_name: &str,
    ) -> Result<(), RuntimeConfigError> {
        let handle = self
            .tcp_server_discovery_runtime
            .get(interface_name)
            .ok_or(RuntimeConfigError {
                code: RuntimeConfigErrorCode::NotFound,
                message: format!("tcp server interface '{}' not found", interface_name),
            })?
            .clone();

        if handle.current.discoverable {
            let iface = Self::make_tcp_server_discoverable_interface(&handle);
            if let Some(announcer) = self.interface_announcer.as_mut() {
                announcer.upsert_interface(iface);
            } else if let Some(identity) = self.transport_identity.as_ref() {
                self.interface_announcer = Some(crate::discovery::InterfaceAnnouncer::new(
                    *identity.hash(),
                    vec![iface],
                ));
            }
        } else if let Some(announcer) = self.interface_announcer.as_mut() {
            announcer.remove_interface(interface_name);
            if announcer.is_empty() {
                self.interface_announcer = None;
            }
        }

        Ok(())
    }

    #[cfg(feature = "hooks")]
    fn update_hook_program<F>(
        &mut self,
        name: &str,
        attach_point: &str,
        mut update: F,
    ) -> Result<(), String>
    where
        F: FnMut(&mut rns_hooks::LoadedProgram),
    {
        let point_idx = crate::config::parse_hook_point(attach_point)
            .ok_or_else(|| format!("unknown hook point '{}'", attach_point))?;
        let program = self.hook_slots[point_idx]
            .programs
            .iter_mut()
            .find(|program| program.name == name)
            .ok_or_else(|| format!("hook '{}' not found at point '{}'", name, attach_point))?;
        update(program);
        Ok(())
    }

    pub(crate) fn set_tick_interval_handle(&mut self, tick_interval_ms: Arc<AtomicU64>) {
        self.tick_interval_ms = tick_interval_ms;
    }

    /// Share the node's stamper, so announces and path requests wait in the
    /// same priority queues as upstream's single stamper thread.
    pub(crate) fn set_stamper(&mut self, stamper: Arc<crate::stamper::Stamper>) {
        self.stamper = stamper;
    }

    /// Apply the `[stamps]` config section and `[reticulum] hop_limiter`,
    /// as `Reticulum.__apply_config` sets the `Transport` attributes before
    /// transport starts (Reticulum.py, d7749b0b). The configured flags also
    /// become the defaults the `global.stamp_*` runtime keys reset to.
    ///
    /// Call after [`Self::set_stamper`] and
    /// [`Self::set_stamp_destinations_flag`], which this updates in place.
    pub(crate) fn apply_stamp_config(
        &mut self,
        stamps: &crate::config::StampsSection,
        hop_limiter: u8,
    ) {
        self.engine.set_stamp_requirements(stamps.requirements());
        self.engine.set_stamp_prs(stamps.stamp_prs);
        self.stamp_destinations
            .store(stamps.stamp_destinations, Ordering::Relaxed);
        self.stamper
            .set_targets(crate::stamper::StampTargets::from(stamps));
        self.stamper
            .set_path_request_timeout(stamps.path_request_stamp_timeout());
        self.runtime_config_defaults.stamp_destinations = stamps.stamp_destinations;
        self.runtime_config_defaults.stamp_path_requests = stamps.stamp_prs;
        self.runtime_config_defaults.stamp_link_requests = stamps.stamp_lrs;
        self.runtime_config_defaults.always_stamp_link_requests = stamps.always_stamp_lrs;
        self.runtime_config_defaults.stamps_enforced = stamps.stamps_enforced;
        self.runtime_config_defaults.lr_stamps_enforced = stamps.lr_stamps_enforced;
        // The per-destination cost table shared with the node API and the
        // announce builder: `lr_cost` is `Transport.LR_COST`, the cost of
        // destinations that call no `set_link_cost`.
        let link_request_costs = self.link_manager.link_request_costs();
        link_request_costs.set_default_cost(stamps.lr_cost);
        link_request_costs.set_enforced(stamps.lr_stamps_enforced);
        self.stamp_settings = stamps.clone();
        self.hop_limiter = hop_limiter;
        // Transport.start (Transport.py:521).
        if hop_limiter != rns_core::constants::PATHFINDER_M {
            log::debug!("Transport hop limit set to {hop_limiter}");
        }
    }

    /// Switch announce and path request stamp enforcement
    /// (`Transport.STAMPS_ENFORCED`), keeping the required costs.
    pub(crate) fn set_stamps_enforced(&mut self, enforced: bool) {
        let mut requirements = self.engine.stamp_requirements();
        requirements.enforced = enforced;
        self.engine.set_stamp_requirements(requirements);
    }

    /// Share the node handle's `STAMP_DESTINATIONS` flag so runtime config
    /// changes made here are seen by the announce API.
    pub(crate) fn set_stamp_destinations_flag(&mut self, flag: Arc<AtomicBool>) {
        self.stamp_destinations = flag;
    }

    /// The flag the node handle reads to refuse new work once draining starts.
    pub(crate) fn accepting_work_handle(&self) -> Arc<AtomicBool> {
        Arc::clone(&self.accepting_work)
    }

    #[allow(dead_code)]
    pub(crate) fn set_packet_hashlist_max_entries(&mut self, max_entries: usize) {
        self.engine.set_packet_hashlist_max_entries(max_entries);
    }

    /// Turn a packet hook chain result into the action the hook point takes.
    ///
    /// `Drop` stops the packet. `Modify` with replacement bytes is validated
    /// against `max_len` and the packet decoder; a replacement that fails
    /// validation is treated as `Drop` and counted in
    /// `hook_rejected_modifications`. `Modify` without bytes, `Continue` and
    /// `Halt` keep the current bytes.
    #[cfg(feature = "hooks")]
    pub(crate) fn packet_hook_outcome(
        &mut self,
        point: HookPoint,
        exec: &rns_hooks::ExecuteResult,
        max_len: usize,
    ) -> PacketHookOutcome {
        let Some(result) = exec.hook_result.as_ref() else {
            return PacketHookOutcome::Keep;
        };
        if result.is_drop() {
            return PacketHookOutcome::Drop;
        }
        if rns_hooks::Verdict::from_u32(result.verdict) != Some(rns_hooks::Verdict::Modify) {
            return PacketHookOutcome::Keep;
        }
        let Some(data) = exec.modified_data.as_deref() else {
            return PacketHookOutcome::Keep;
        };
        match validate_hook_replacement(data, max_len) {
            Ok(packet) => PacketHookOutcome::Replace(packet),
            Err(reason) => {
                let rejected = &mut self.hook_rejected_modifications[point as usize];
                *rejected = rejected.saturating_add(1);
                log::debug!("{:?} hook: {}; dropping packet", point, reason);
                PacketHookOutcome::Drop
            }
        }
    }

    /// Largest packet a hook may substitute for one handled on `interface`.
    #[cfg(feature = "hooks")]
    pub(crate) fn hook_packet_limit(&self, interface: InterfaceId) -> usize {
        self.interfaces
            .get(&interface)
            .map(|entry| entry.info.mtu as usize)
            .filter(|mtu| *mtu > 0)
            .unwrap_or(rns_core::constants::MTU)
    }

    /// Point a pending confirmed link send at the replacement bytes a hook
    /// produced for the packet it was tracking.
    #[cfg(feature = "hooks")]
    pub(crate) fn retarget_tracked_link_send(&mut self, original: &[u8], replacement: &RawPacket) {
        let Some((hash, _)) = self.tracked_link_send.as_mut() else {
            return;
        };
        if RawPacket::unpack(original).is_ok_and(|packet| packet.packet_hash == *hash) {
            *hash = replacement.packet_hash;
        }
    }

    #[cfg(feature = "hooks")]
    fn forward_hook_side_effects(&mut self, attach_point: &str, exec: &rns_hooks::ExecuteResult) {
        if !exec.injected_actions.is_empty() {
            self.dispatch_all(convert_injected_actions(exec.injected_actions.clone()));
        }
        if let Some(ref bridge) = self.provider_bridge {
            for event in &exec.provider_events {
                bridge.emit_event(
                    attach_point,
                    event.hook_name.clone(),
                    event.payload_type.clone(),
                    event.payload.clone(),
                );
            }
        }
    }

    #[cfg(feature = "hooks")]
    fn collect_hook_side_effects(
        &mut self,
        attach_point: &str,
        exec: &rns_hooks::ExecuteResult,
        out: &mut Vec<TransportAction>,
    ) {
        if !exec.injected_actions.is_empty() {
            out.extend(convert_injected_actions(exec.injected_actions.clone()));
        }
        if let Some(ref bridge) = self.provider_bridge {
            for event in &exec.provider_events {
                bridge.emit_event(
                    attach_point,
                    event.hook_name.clone(),
                    event.payload_type.clone(),
                    event.payload.clone(),
                );
            }
        }
    }

    /// Set the probe addresses, protocol, and optional device for hole punching.
    pub fn set_probe_config(
        &mut self,
        addrs: Vec<std::net::SocketAddr>,
        protocol: rns_core::holepunch::ProbeProtocol,
        device: Option<String>,
    ) {
        let policy = self.holepunch_manager.policy();
        let allow_private = self.holepunch_manager.allow_private_endpoints();
        self.holepunch_manager = HolePunchManager::new(addrs, protocol, device);
        self.holepunch_manager
            .set_event_sender(self.event_tx.clone());
        self.holepunch_manager.set_underlay_mark(self.underlay_mark);
        self.holepunch_manager.set_policy(policy);
        self.holepunch_manager
            .set_allow_private_endpoints(allow_private);
    }
}

pub(crate) struct PendingLinkFrame {
    interface: InterfaceId,
    data: Vec<u8>,
    is_announce: bool,
    completion: Option<crate::link_send::Completion>,
}

/// Wakes a worker thread when there is work for it, or when it should
/// stop, so it sleeps until then instead of polling.
#[derive(Default)]
pub(crate) struct WorkSignal {
    /// Work is pending; the worker is stopped.
    state: Mutex<(bool, bool)>,
    condvar: std::sync::Condvar,
}

impl WorkSignal {
    fn lock(&self) -> std::sync::MutexGuard<'_, (bool, bool)> {
        self.state
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// There is work.
    pub(crate) fn notify(&self) {
        self.lock().0 = true;
        self.condvar.notify_one();
    }

    /// The worker should stop.
    pub(crate) fn stop(&self) {
        self.lock().1 = true;
        self.condvar.notify_all();
    }

    pub(crate) fn stopped(&self) -> bool {
        self.lock().1
    }

    /// Wait until there is work or the worker should stop, consuming the
    /// notification. Returns `false` once stopped.
    pub(crate) fn wait(&self) -> bool {
        let mut state = self.lock();
        while !state.0 && !state.1 {
            state = self
                .condvar
                .wait(state)
                .unwrap_or_else(|poisoned| poisoned.into_inner());
        }
        state.0 = false;
        !state.1
    }
}