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) = crate::upper::tun::tun_outbound_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        self.attach_endpoint_data_io_inner(capacity, None)
53    }
54
55    pub(crate) fn attach_endpoint_data_io_with_direct_sink(
56        &mut self,
57        capacity: usize,
58        direct_sink: EndpointDirectSink,
59    ) -> Result<EndpointDataIo, NodeError> {
60        self.attach_endpoint_data_io_inner(capacity, Some(direct_sink))
61    }
62
63    fn attach_endpoint_data_io_inner(
64        &mut self,
65        capacity: usize,
66        direct_sink: Option<EndpointDirectSink>,
67    ) -> Result<EndpointDataIo, NodeError> {
68        if self.state != NodeState::Created {
69            return Err(NodeError::Config(ConfigError::Validation(
70                "endpoint data I/O must be attached before node start".to_string(),
71            )));
72        }
73
74        let command_capacity = capacity.max(1);
75        let (control_tx, control_rx) = tokio::sync::mpsc::channel(command_capacity);
76        let (data_batch_tx, data_rx) = endpoint_data_batch_channel(command_capacity);
77        // Endpoint events use one bounded app-data channel. Protocol/control
78        // progress is reserved before endpoint payload delivery reaches this
79        // queue.
80        let (event_tx, event_rx) =
81            EndpointEventSender::channel_with_direct_sink(capacity, direct_sink);
82        let (service_event_tx, service_event_rx) = EndpointServiceEventSender::channel(capacity);
83        self.endpoint_control_rx = Some(control_rx);
84        self.endpoint_data_rx = Some(data_rx);
85        self.endpoint_events.attach(event_tx.clone());
86        self.endpoint_services.attach(service_event_tx.clone());
87
88        Ok(EndpointDataIo {
89            control_tx,
90            data_batch_tx,
91            event_rx,
92            event_tx,
93            service_event_rx,
94            service_event_tx,
95        })
96    }
97
98    pub(crate) fn pubkey_for_node_addr(&self, addr: &NodeAddr) -> Option<secp256k1::PublicKey> {
99        self.identity_cache.pubkey_for_node_addr(addr)
100    }
101
102    pub(crate) fn npub_for_node_addr(&self, addr: &NodeAddr) -> Option<String> {
103        self.identity_cache.npub_for_node_addr(addr)
104    }
105
106    pub(in crate::node) fn deliver_external_ipv6_packet(
107        &self,
108        src_addr: &NodeAddr,
109        packet: Vec<u8>,
110    ) {
111        let Some(external_packet_tx) = &self.external_packet_tx else {
112            return;
113        };
114        if packet.len() < 40 {
115            return;
116        }
117        let Ok(destination) = FipsAddress::from_slice(&packet[24..40]) else {
118            return;
119        };
120        let delivered = NodeDeliveredPacket {
121            source_node_addr: *src_addr,
122            source_npub: self.npub_for_node_addr(src_addr),
123            destination,
124            packet,
125        };
126        if let Err(error) = external_packet_tx.try_send(delivered) {
127            debug!(error = %error, "Failed to deliver packet to external app sink");
128        }
129    }
130
131    /// Update one peer's local-outbound-broken signal from a `transport.send`
132    /// outcome. Sets a per-peer timestamp on local-side io errors
133    /// (NetworkUnreachable / HostUnreachable / AddrNotAvailable); clears that
134    /// peer on success. The reaper consults this in `check_link_heartbeats` to
135    /// switch only that peer to `fast_link_dead_timeout_secs`.
136    pub(in crate::node) fn note_local_send_outcome(
137        &mut self,
138        node_addr: &NodeAddr,
139        result: &Result<usize, TransportError>,
140    ) {
141        self.local_send_failures
142            .note_send_outcome(node_addr, result, std::time::Instant::now());
143    }
144
145    /// Update local-outbound liveness from a candidate/probe send.
146    ///
147    /// A failed probe to an alternate stale address must not compress the
148    /// live path's dead timeout. Only the active current path, or a peer with
149    /// no usable current path, gets the fast-dead signal.
150    pub(in crate::node) fn note_candidate_send_outcome(
151        &mut self,
152        node_addr: &NodeAddr,
153        candidate_addr: &TransportAddr,
154        result: &Result<usize, TransportError>,
155    ) {
156        if result.is_ok() {
157            self.note_local_send_outcome(node_addr, result);
158            return;
159        }
160
161        let candidate_is_active_path = self
162            .peers
163            .get(node_addr)
164            .is_none_or(|peer| !peer.can_send() || peer.current_addr() == Some(candidate_addr));
165        if candidate_is_active_path {
166            self.note_local_send_outcome(node_addr, result);
167        }
168    }
169
170    /// Return the active dead-timeout for one peer after considering recent
171    /// local route failures. The fast-dead signal is intentionally short-lived:
172    /// on the UDP worker path a send call can return before the kernel result
173    /// is observed, so a stale route error must not compress liveness for the
174    /// whole normal dead-timeout window.
175    pub(in crate::node) fn local_send_failure_dead_timeout_for_peer(
176        &self,
177        node_addr: &NodeAddr,
178        now: std::time::Instant,
179        dead_timeout: std::time::Duration,
180        fast_dead_timeout: std::time::Duration,
181    ) -> std::time::Duration {
182        self.local_send_failures.dead_timeout_for_peer(
183            node_addr,
184            now,
185            dead_timeout,
186            fast_dead_timeout,
187        )
188    }
189
190    pub(in crate::node) fn purge_expired_local_send_failures(&mut self, now: std::time::Instant) {
191        self.local_send_failures.purge_expired(now);
192    }
193
194    pub(in crate::node) fn mark_rx_loop_maintenance_timeout(&mut self) {
195        self.last_rx_loop_maintenance_timeout_at = Some(std::time::Instant::now());
196    }
197
198    pub(in crate::node) fn rx_loop_maintenance_timed_out_recently(&self) -> bool {
199        let Some(t) = self.last_rx_loop_maintenance_timeout_at else {
200            return false;
201        };
202        let grace = std::time::Duration::from_secs(self.config.node.link_dead_timeout_secs.max(1));
203        std::time::Instant::now().duration_since(t) <= grace
204    }
205}
fips_core/node/io_impl.rs

fips_core/node/
io_impl.rs
