fips_core/node/
io_impl.rs

1use super::*;
2
3impl Node {
4    /// Attach application-owned I/O for the Nostr relay fallback transport.
5    ///
6    /// The returned handle is safe to use while [`Node::run_rx_loop`] owns the
7    /// node. Relay selection and connections remain in the embedding
8    /// application. This must be called before [`Node::start`].
9    pub fn attach_nostr_relay_io(
10        &mut self,
11        capacity: usize,
12    ) -> Result<crate::endpoint::NostrRelayIo, NodeError> {
13        if self.state != NodeState::Created {
14            return Err(NodeError::Config(ConfigError::Validation(
15                "Nostr relay I/O must be attached before node start".to_string(),
16            )));
17        }
18        if self.endpoint_control_rx.is_some() {
19            return Err(NodeError::Config(ConfigError::Validation(
20                "node application control I/O is already attached".to_string(),
21            )));
22        }
23
24        let (control_tx, control_rx) = tokio::sync::mpsc::channel(capacity.max(1));
25        self.endpoint_control_rx = Some(control_rx);
26        Ok(crate::endpoint::NostrRelayIo::new(
27            self.identity.npub(),
28            control_tx,
29        ))
30    }
31
32    /// Get the TUN packet sender channel.
33    ///
34    /// Returns None if TUN is not active or the node hasn't been started.
35    pub fn tun_tx(&self) -> Option<&TunTx> {
36        self.tun_tx.as_ref()
37    }
38
39    /// Attach app-owned packet I/O for embedded operation without a system TUN.
40    ///
41    /// This must be called before [`Node::start`] and requires `tun.enabled =
42    /// false`. Outbound packets sent to the returned sender are processed by the
43    /// normal session pipeline. Inbound packets delivered by FIPS sessions are
44    /// sent to the returned receiver with source attribution.
45    pub fn attach_external_packet_io(
46        &mut self,
47        capacity: usize,
48    ) -> Result<ExternalPacketIo, NodeError> {
49        if self.state != NodeState::Created {
50            return Err(NodeError::Config(ConfigError::Validation(
51                "external packet I/O must be attached before node start".to_string(),
52            )));
53        }
54        if self.config.tun.enabled {
55            return Err(NodeError::Config(ConfigError::Validation(
56                "external packet I/O requires tun.enabled=false".to_string(),
57            )));
58        }
59
60        let capacity = capacity.max(1);
61        let (outbound_tx, outbound_rx) = crate::upper::tun::tun_outbound_channel(capacity);
62        let (inbound_tx, inbound_rx) = tokio::sync::mpsc::channel(capacity);
63        self.tun_outbound_rx = Some(outbound_rx);
64        self.external_packet_tx = Some(inbound_tx);
65
66        Ok(ExternalPacketIo {
67            outbound_tx,
68            inbound_rx,
69        })
70    }
71
72    /// Attach app-owned endpoint data I/O for embedded operation.
73    ///
74    /// Commands sent to the returned sender are processed by the node RX loop.
75    /// Incoming endpoint data is emitted as source-attributed events.
76    pub(crate) fn attach_endpoint_data_io(
77        &mut self,
78        capacity: usize,
79    ) -> Result<EndpointDataIo, NodeError> {
80        self.attach_endpoint_data_io_inner(capacity, None)
81    }
82
83    pub(crate) fn attach_endpoint_data_io_with_direct_sink(
84        &mut self,
85        capacity: usize,
86        direct_sink: EndpointDirectSink,
87    ) -> Result<EndpointDataIo, NodeError> {
88        self.attach_endpoint_data_io_inner(capacity, Some(direct_sink))
89    }
90
91    fn attach_endpoint_data_io_inner(
92        &mut self,
93        capacity: usize,
94        direct_sink: Option<EndpointDirectSink>,
95    ) -> Result<EndpointDataIo, NodeError> {
96        if self.state != NodeState::Created {
97            return Err(NodeError::Config(ConfigError::Validation(
98                "endpoint data I/O must be attached before node start".to_string(),
99            )));
100        }
101
102        let command_capacity = capacity.max(1);
103        let (control_tx, control_rx) = tokio::sync::mpsc::channel(command_capacity);
104        let (data_batch_tx, data_rx) = endpoint_data_batch_channel(command_capacity);
105        // Endpoint events use one bounded app-data channel. Protocol/control
106        // progress is reserved before endpoint payload delivery reaches this
107        // queue.
108        let (event_tx, event_rx) =
109            EndpointEventSender::channel_with_direct_sink(capacity, direct_sink);
110        let (service_event_tx, service_event_rx) = EndpointServiceEventSender::channel(capacity);
111        self.endpoint_control_rx = Some(control_rx);
112        self.endpoint_data_rx = Some(data_rx);
113        self.endpoint_events.attach(event_tx.clone());
114
115        Ok(EndpointDataIo {
116            control_tx,
117            data_batch_tx,
118            event_rx,
119            event_tx,
120            service_event_rx,
121            service_event_tx,
122        })
123    }
124
125    pub(crate) fn pubkey_for_node_addr(&self, addr: &NodeAddr) -> Option<secp256k1::PublicKey> {
126        self.identity_cache.pubkey_for_node_addr(addr)
127    }
128
129    pub(crate) fn npub_for_node_addr(&self, addr: &NodeAddr) -> Option<String> {
130        self.identity_cache.npub_for_node_addr(addr)
131    }
132
133    pub(in crate::node) fn deliver_external_ipv6_packet(
134        &self,
135        src_addr: &NodeAddr,
136        packet: Vec<u8>,
137    ) {
138        let Some(external_packet_tx) = &self.external_packet_tx else {
139            return;
140        };
141        if packet.len() < 40 {
142            return;
143        }
144        let Ok(destination) = FipsAddress::from_slice(&packet[24..40]) else {
145            return;
146        };
147        let delivered = NodeDeliveredPacket {
148            source_node_addr: *src_addr,
149            source_npub: self.npub_for_node_addr(src_addr),
150            destination,
151            packet,
152        };
153        if let Err(error) = external_packet_tx.try_send(delivered) {
154            debug!(error = %error, "Failed to deliver packet to external app sink");
155        }
156    }
157
158    /// Update one peer's local-outbound-broken signal from a `transport.send`
159    /// outcome. Sets a per-peer timestamp on local-side io errors
160    /// (NetworkUnreachable / HostUnreachable / AddrNotAvailable); clears that
161    /// peer on success. The reaper consults this in `check_link_heartbeats` to
162    /// switch only that peer to `fast_link_dead_timeout_secs`.
163    pub(in crate::node) fn note_local_send_outcome(
164        &mut self,
165        node_addr: &NodeAddr,
166        result: &Result<usize, TransportError>,
167    ) {
168        self.local_send_failures
169            .note_send_outcome(node_addr, result, std::time::Instant::now());
170    }
171
172    /// Update local-outbound liveness from a candidate/probe send.
173    ///
174    /// A failed probe to an alternate stale address must not compress the
175    /// live path's dead timeout. Only the active current path, or a peer with
176    /// no usable current path, gets the fast-dead signal.
177    pub(in crate::node) fn note_candidate_send_outcome(
178        &mut self,
179        node_addr: &NodeAddr,
180        candidate_addr: &TransportAddr,
181        result: &Result<usize, TransportError>,
182    ) {
183        if result.is_ok() {
184            self.note_local_send_outcome(node_addr, result);
185            return;
186        }
187
188        let candidate_is_active_path = self
189            .peers
190            .get(node_addr)
191            .is_none_or(|peer| !peer.can_send() || peer.current_addr() == Some(candidate_addr));
192        if candidate_is_active_path {
193            self.note_local_send_outcome(node_addr, result);
194        }
195    }
196
197    /// Return the active dead-timeout for one peer after considering recent
198    /// local route failures. The fast-dead signal is intentionally short-lived:
199    /// on the UDP worker path a send call can return before the kernel result
200    /// is observed, so a stale route error must not compress liveness for the
201    /// whole normal dead-timeout window.
202    pub(in crate::node) fn local_send_failure_dead_timeout_for_peer(
203        &self,
204        node_addr: &NodeAddr,
205        now: std::time::Instant,
206        dead_timeout: std::time::Duration,
207        fast_dead_timeout: std::time::Duration,
208    ) -> std::time::Duration {
209        self.local_send_failures.dead_timeout_for_peer(
210            node_addr,
211            now,
212            dead_timeout,
213            fast_dead_timeout,
214        )
215    }
216
217    pub(in crate::node) fn purge_expired_local_send_failures(&mut self, now: std::time::Instant) {
218        self.local_send_failures.purge_expired(now);
219    }
220
221    pub(in crate::node) fn mark_rx_loop_maintenance_timeout(&mut self) {
222        let now = std::time::Instant::now();
223        let grace = std::time::Duration::from_secs(self.config.node.link_dead_timeout_secs.max(1));
224        if self
225            .last_rx_loop_maintenance_timeout_at
226            .is_some_and(|started| now.duration_since(started) <= grace)
227        {
228            return;
229        }
230        self.last_rx_loop_maintenance_timeout_at = Some(now);
231    }
232
233    pub(in crate::node) fn rx_loop_maintenance_timed_out_recently(&self) -> bool {
234        let Some(t) = self.last_rx_loop_maintenance_timeout_at else {
235            return false;
236        };
237        let grace = std::time::Duration::from_secs(self.config.node.link_dead_timeout_secs.max(1));
238        std::time::Instant::now().duration_since(t) <= grace
239    }
240}
fips_core/node/io_impl.rs

fips_core/node/
io_impl.rs
