fips_core/node/
io_impl.rs

1use super::*;
2
3impl Node {
4    /// Get the TUN packet sender channel.
5    ///
6    /// Returns None if TUN is not active or the node hasn't been started.
7    pub fn tun_tx(&self) -> Option<&TunTx> {
8        self.tun_tx.as_ref()
9    }
10
11    /// Attach app-owned packet I/O for embedded operation without a system TUN.
12    ///
13    /// This must be called before [`Node::start`] and requires `tun.enabled =
14    /// false`. Outbound packets sent to the returned sender are processed by the
15    /// normal session pipeline. Inbound packets delivered by FIPS sessions are
16    /// sent to the returned receiver with source attribution.
17    pub fn attach_external_packet_io(
18        &mut self,
19        capacity: usize,
20    ) -> Result<ExternalPacketIo, NodeError> {
21        if self.state != NodeState::Created {
22            return Err(NodeError::Config(ConfigError::Validation(
23                "external packet I/O must be attached before node start".to_string(),
24            )));
25        }
26        if self.config.tun.enabled {
27            return Err(NodeError::Config(ConfigError::Validation(
28                "external packet I/O requires tun.enabled=false".to_string(),
29            )));
30        }
31
32        let capacity = capacity.max(1);
33        let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(capacity);
34        let (inbound_tx, inbound_rx) = tokio::sync::mpsc::channel(capacity);
35        self.tun_outbound_rx = Some(outbound_rx);
36        self.external_packet_tx = Some(inbound_tx);
37
38        Ok(ExternalPacketIo {
39            outbound_tx,
40            inbound_rx,
41        })
42    }
43
44    /// Attach app-owned endpoint data I/O for embedded operation.
45    ///
46    /// Commands sent to the returned sender are processed by the node RX loop.
47    /// Incoming endpoint data is emitted as source-attributed events.
48    pub(crate) fn attach_endpoint_data_io(
49        &mut self,
50        capacity: usize,
51    ) -> Result<EndpointDataIo, NodeError> {
52        if self.state != NodeState::Created {
53            return Err(NodeError::Config(ConfigError::Validation(
54                "endpoint data I/O must be attached before node start".to_string(),
55            )));
56        }
57
58        let command_capacity = endpoint_data_command_capacity(capacity);
59        let (priority_command_tx, priority_command_rx) =
60            tokio::sync::mpsc::channel(command_capacity);
61        let (command_tx, command_rx) = tokio::sync::mpsc::channel(command_capacity);
62        // Endpoint events keep priority delivery wait-free and bound bulk
63        // backlog by the caller's packet-channel capacity.
64        let (event_tx, event_rx) = EndpointEventSender::channel(capacity);
65        self.endpoint_priority_command_rx = Some(priority_command_rx);
66        self.endpoint_command_rx = Some(command_rx);
67        self.endpoint_events.attach(event_tx.clone());
68
69        Ok(EndpointDataIo {
70            priority_command_tx,
71            command_tx,
72            event_rx,
73            event_tx,
74        })
75    }
76
77    pub(in crate::node) fn begin_endpoint_event_batch(&mut self) {
78        self.endpoint_events.begin_batch();
79    }
80
81    pub(in crate::node) fn finish_endpoint_event_batch(&mut self) {
82        self.endpoint_events.finish_batch();
83    }
84
85    #[allow(clippy::result_large_err)]
86    pub(in crate::node) fn deliver_endpoint_event_message(
87        &mut self,
88        message: EndpointDataDelivery,
89    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
90        self.endpoint_events.deliver_endpoint_data(message)
91    }
92
93    pub(in crate::node) fn decrypt_direct_session_delivery_sink(
94        &self,
95    ) -> decrypt_worker::DecryptDirectSessionDeliverySink {
96        decrypt_worker::DecryptDirectSessionDeliverySink::new(
97            self.tun_tx.clone(),
98            self.external_packet_tx.clone(),
99            self.endpoint_events.sender(),
100        )
101    }
102
103    pub(crate) fn pubkey_for_node_addr(&self, addr: &NodeAddr) -> Option<secp256k1::PublicKey> {
104        self.identity_cache.pubkey_for_node_addr(addr)
105    }
106
107    pub(crate) fn npub_for_node_addr(&self, addr: &NodeAddr) -> Option<String> {
108        self.identity_cache.npub_for_node_addr(addr)
109    }
110
111    pub(in crate::node) fn deliver_external_ipv6_packet(
112        &self,
113        src_addr: &NodeAddr,
114        packet: Vec<u8>,
115    ) {
116        let Some(external_packet_tx) = &self.external_packet_tx else {
117            return;
118        };
119        if packet.len() < 40 {
120            return;
121        }
122        let Ok(destination) = FipsAddress::from_slice(&packet[24..40]) else {
123            return;
124        };
125        let delivered = NodeDeliveredPacket {
126            source_node_addr: *src_addr,
127            source_npub: self.npub_for_node_addr(src_addr),
128            destination,
129            packet,
130        };
131        if let Err(error) = external_packet_tx.try_send(delivered) {
132            debug!(error = %error, "Failed to deliver packet to external app sink");
133        }
134    }
135
136    /// Update one peer's local-outbound-broken signal from a `transport.send`
137    /// outcome. Sets a per-peer timestamp on local-side io errors
138    /// (NetworkUnreachable / HostUnreachable / AddrNotAvailable); clears that
139    /// peer on success. The reaper consults this in `check_link_heartbeats` to
140    /// switch only that peer to `fast_link_dead_timeout_secs`.
141    pub(in crate::node) fn note_local_send_outcome(
142        &mut self,
143        node_addr: &NodeAddr,
144        result: &Result<usize, TransportError>,
145    ) {
146        self.local_send_failures
147            .note_send_outcome(node_addr, result, std::time::Instant::now());
148    }
149
150    /// Return the active dead-timeout for one peer after considering recent
151    /// local route failures. The fast-dead signal is intentionally short-lived:
152    /// on the UDP worker path a send call can return before the kernel result
153    /// is observed, so a stale route error must not compress liveness for the
154    /// whole normal dead-timeout window.
155    pub(in crate::node) fn local_send_failure_dead_timeout_for_peer(
156        &self,
157        node_addr: &NodeAddr,
158        now: std::time::Instant,
159        dead_timeout: std::time::Duration,
160        fast_dead_timeout: std::time::Duration,
161    ) -> std::time::Duration {
162        self.local_send_failures.dead_timeout_for_peer(
163            node_addr,
164            now,
165            dead_timeout,
166            fast_dead_timeout,
167        )
168    }
169
170    pub(in crate::node) fn purge_expired_local_send_failures(&mut self, now: std::time::Instant) {
171        self.local_send_failures.purge_expired(now);
172    }
173
174    pub(in crate::node) fn mark_rx_loop_maintenance_timeout(&mut self) {
175        self.last_rx_loop_maintenance_timeout_at = Some(std::time::Instant::now());
176    }
177
178    pub(in crate::node) fn rx_loop_maintenance_timed_out_recently(&self) -> bool {
179        let Some(t) = self.last_rx_loop_maintenance_timeout_at else {
180            return false;
181        };
182        let grace = std::time::Duration::from_secs(self.config.node.link_dead_timeout_secs.max(1));
183        std::time::Instant::now().duration_since(t) <= grace
184    }
185}
fips_core/node/io_impl.rs

fips_core/node/
io_impl.rs
