fips_core/node/
endpoint_event.rs

1use super::*;
2use crate::transport::PacketBuffer;
3use std::collections::VecDeque;
4use std::sync::{Arc, Condvar, Mutex};
5use std::time::Duration;
6
7/// Maximum endpoint packets in one authenticated direct packet run.
8pub const FIPS_ENDPOINT_DIRECT_PACKET_RUN_MAX_PACKETS: usize = 128;
9
10/// Maximum endpoint packets pending in an installed direct packet sink.
11pub const FIPS_ENDPOINT_DIRECT_PACKET_QUEUE_MAX_PACKETS: usize = 4096;
12
13/// Authenticated source/session facts for a direct endpoint packet run.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub(crate) struct FipsEndpointDirectPacketRunMeta {
16    source_peer: PeerIdentity,
17    previous_hop_addr: NodeAddr,
18    received_k_bit: bool,
19    direct_path: bool,
20    enqueued_at_ms: u64,
21}
22
23impl FipsEndpointDirectPacketRunMeta {
24    pub(crate) fn new(
25        source_peer: PeerIdentity,
26        previous_hop_addr: NodeAddr,
27        received_k_bit: bool,
28        direct_path: bool,
29        enqueued_at_ms: u64,
30    ) -> Self {
31        Self {
32            source_peer,
33            previous_hop_addr,
34            received_k_bit,
35            direct_path,
36            enqueued_at_ms,
37        }
38    }
39}
40
41/// Consecutive direct endpoint packets from one authenticated FIPS source.
42///
43/// FIPS owns authentication and ordering. The embedder can still apply live
44/// routing policy before borrowing packet bytes for TUN writes.
45#[derive(Debug, Clone, PartialEq, Eq)]
46pub struct FipsEndpointDirectPacketRun {
47    meta: FipsEndpointDirectPacketRunMeta,
48    packets: Vec<PacketBuffer>,
49    packet_bytes: usize,
50}
51
52/// Borrowed packet slices from a direct endpoint packet run.
53pub struct FipsEndpointDirectPacketSlices<'a> {
54    packets: std::slice::Iter<'a, PacketBuffer>,
55}
56
57impl FipsEndpointDirectPacketRun {
58    pub(crate) fn from_packet(meta: FipsEndpointDirectPacketRunMeta, packet: PacketBuffer) -> Self {
59        let packet_bytes = packet.len();
60        let mut packets = Vec::with_capacity(FIPS_ENDPOINT_DIRECT_PACKET_RUN_MAX_PACKETS);
61        packets.push(packet);
62        Self {
63            meta,
64            packets,
65            packet_bytes,
66        }
67    }
68
69    /// Authenticated FIPS peer that originated every packet in this run.
70    pub fn source_peer(&self) -> &PeerIdentity {
71        &self.meta.source_peer
72    }
73
74    /// Unix-millisecond time when FIPS handed this run to the direct sink.
75    pub fn enqueued_at_ms(&self) -> u64 {
76        self.meta.enqueued_at_ms
77    }
78
79    /// Number of endpoint packets in the run.
80    pub fn len(&self) -> usize {
81        self.packets.len()
82    }
83
84    /// Whether the run contains no packets.
85    pub fn is_empty(&self) -> bool {
86        self.len() == 0
87    }
88
89    /// Sum of endpoint packet bytes, excluding bulk length metadata.
90    pub fn packet_bytes(&self) -> usize {
91        self.packet_bytes
92    }
93
94    /// Borrow one packet by index.
95    pub fn packet_slice(&self, index: usize) -> Option<&[u8]> {
96        self.packets.get(index).map(PacketBuffer::as_slice)
97    }
98
99    pub(crate) fn push_packet(&mut self, packet: PacketBuffer) {
100        self.packet_bytes = self.packet_bytes.saturating_add(packet.len());
101        self.packets.push(packet);
102    }
103
104    pub(crate) fn try_append(&mut self, other: &mut Self) -> bool {
105        if self.len().saturating_add(other.len()) > FIPS_ENDPOINT_DIRECT_PACKET_RUN_MAX_PACKETS
106            || self.meta.source_peer != other.meta.source_peer
107            || self.meta.previous_hop_addr != other.meta.previous_hop_addr
108            || self.meta.received_k_bit != other.meta.received_k_bit
109            || self.meta.direct_path != other.meta.direct_path
110        {
111            return false;
112        }
113        self.packet_bytes = self.packet_bytes.saturating_add(other.packet_bytes);
114        other.packet_bytes = 0;
115        self.packets.append(&mut other.packets);
116        true
117    }
118
119    /// Borrow packet bytes from the run-owned buffers.
120    pub fn packet_slices(&self) -> FipsEndpointDirectPacketSlices<'_> {
121        FipsEndpointDirectPacketSlices {
122            packets: self.packets.iter(),
123        }
124    }
125
126    /// Keep only packets accepted by the caller while preserving packet buffers.
127    ///
128    /// The predicate receives the original packet index and immutable bytes. This
129    /// keeps routing/admission policy outside FIPS while allowing embedders to
130    /// remove rejected packets before a TUN writer borrows the run.
131    pub fn retain_packets<F>(&mut self, mut keep: F)
132    where
133        F: FnMut(usize, &[u8]) -> bool,
134    {
135        let mut index = 0usize;
136        let mut packet_bytes = 0usize;
137        self.packets.retain(|packet| {
138            let retained = keep(index, packet.as_slice());
139            index = index.saturating_add(1);
140            if retained {
141                packet_bytes = packet_bytes.saturating_add(packet.len());
142            }
143            retained
144        });
145        self.packet_bytes = packet_bytes;
146    }
147
148    /// Split this run at a packet index without copying packet bytes.
149    ///
150    /// The original run keeps packets before `at`; the returned run contains
151    /// packets from `at` onward with the same authenticated source metadata.
152    pub fn split_off_packets(&mut self, at: usize) -> Option<Self> {
153        if at >= self.packets.len() {
154            return None;
155        }
156        let packets = self.packets.split_off(at);
157        let packet_bytes = packets.iter().map(PacketBuffer::len).sum();
158        self.packet_bytes = self.packet_bytes.saturating_sub(packet_bytes);
159        Some(Self {
160            meta: self.meta.clone(),
161            packets,
162            packet_bytes,
163        })
164    }
165}
166
167impl<'a> Iterator for FipsEndpointDirectPacketSlices<'a> {
168    type Item = &'a [u8];
169
170    fn next(&mut self) -> Option<Self::Item> {
171        self.packets.next().map(PacketBuffer::as_slice)
172    }
173
174    fn size_hint(&self) -> (usize, Option<usize>) {
175        self.packets.size_hint()
176    }
177}
178
179impl ExactSizeIterator for FipsEndpointDirectPacketSlices<'_> {}
180
181impl Drop for FipsEndpointDirectPacketRun {
182    fn drop(&mut self) {
183        PacketBuffer::recycle_batch(&mut self.packets);
184    }
185}
186
187/// Established endpoint packet runs delivered without the endpoint-event queue.
188#[derive(Debug, Clone, PartialEq, Eq)]
189pub struct FipsEndpointDirectPacketBatch {
190    packet_runs: Vec<FipsEndpointDirectPacketRun>,
191}
192
193impl FipsEndpointDirectPacketBatch {
194    pub(crate) fn from_packet_runs(packet_runs: Vec<FipsEndpointDirectPacketRun>) -> Self {
195        debug_assert!(
196            packet_runs
197                .iter()
198                .all(|run| run.len() <= FIPS_ENDPOINT_DIRECT_PACKET_RUN_MAX_PACKETS)
199        );
200        Self { packet_runs }
201    }
202
203    /// Take ownership of the delivered packet runs.
204    pub fn into_packet_runs(self) -> Vec<FipsEndpointDirectPacketRun> {
205        self.packet_runs
206    }
207}
208
209/// Error returned by an installed direct endpoint sink.
210#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
211pub enum FipsEndpointDirectDeliveryError {
212    /// The sink could not accept this batch.
213    #[error("direct endpoint sink unavailable")]
214    Unavailable,
215}
216
217/// Application-provided direct dataplane endpoint delivery sink.
218///
219/// This sink is called synchronously from the dataplane output path with owned packet
220/// buffers. It should return quickly and avoid blocking unrelated dataplane progress.
221pub trait FipsEndpointDirectSink: Send + Sync + 'static {
222    /// Deliver established endpoint data as authenticated packet runs.
223    fn deliver_endpoint_packet_batch(
224        &self,
225        batch: FipsEndpointDirectPacketBatch,
226    ) -> Result<(), FipsEndpointDirectDeliveryError>;
227}
228
229impl<F> FipsEndpointDirectSink for F
230where
231    F: Fn(FipsEndpointDirectPacketBatch) -> Result<(), FipsEndpointDirectDeliveryError>
232        + Send
233        + Sync
234        + 'static,
235{
236    fn deliver_endpoint_packet_batch(
237        &self,
238        batch: FipsEndpointDirectPacketBatch,
239    ) -> Result<(), FipsEndpointDirectDeliveryError> {
240        self(batch)
241    }
242}
243
244#[derive(Clone)]
245pub(crate) struct EndpointDirectSink {
246    sink: Arc<dyn FipsEndpointDirectSink>,
247}
248
249impl std::fmt::Debug for EndpointDirectSink {
250    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
251        f.debug_struct("EndpointDirectSink").finish_non_exhaustive()
252    }
253}
254
255impl EndpointDirectSink {
256    pub(crate) fn new<S>(sink: S) -> Self
257    where
258        S: FipsEndpointDirectSink,
259    {
260        Self {
261            sink: Arc::new(sink),
262        }
263    }
264
265    pub(crate) fn deliver_direct_packet_batch(
266        &self,
267        batch: FipsEndpointDirectPacketBatch,
268    ) -> Result<(), FipsEndpointDirectDeliveryError> {
269        self.sink.deliver_endpoint_packet_batch(batch)
270    }
271}
272
273struct FipsEndpointDirectReceiverSink {
274    shared: Arc<FipsEndpointDirectReceiverShared>,
275}
276
277/// Blocking bounded receiver for authenticated direct endpoint packet runs.
278#[derive(Debug)]
279pub struct FipsEndpointDirectReceiver {
280    shared: Arc<FipsEndpointDirectReceiverShared>,
281}
282
283#[derive(Debug)]
284struct FipsEndpointDirectReceiverShared {
285    state: Mutex<FipsEndpointDirectReceiverState>,
286    ready: Condvar,
287}
288
289#[derive(Debug, Default)]
290struct FipsEndpointDirectReceiverState {
291    runs: VecDeque<FipsEndpointDirectPacketRun>,
292    packets: usize,
293    interrupted: bool,
294}
295
296impl FipsEndpointDirectReceiver {
297    pub(crate) fn channel() -> (impl FipsEndpointDirectSink, Self) {
298        let shared = Arc::new(FipsEndpointDirectReceiverShared {
299            state: Mutex::new(FipsEndpointDirectReceiverState::default()),
300            ready: Condvar::new(),
301        });
302        (
303            FipsEndpointDirectReceiverSink {
304                shared: Arc::clone(&shared),
305            },
306            Self { shared },
307        )
308    }
309
310    /// Wait for a source-contiguous packet run batch bounded by `packet_limit`.
311    pub fn recv_timeout(
312        &self,
313        timeout: Duration,
314        packet_limit: usize,
315    ) -> Result<Vec<FipsEndpointDirectPacketRun>, std::sync::mpsc::RecvTimeoutError> {
316        let mut state = self
317            .shared
318            .state
319            .lock()
320            .map_err(|_| std::sync::mpsc::RecvTimeoutError::Disconnected)?;
321        if state.runs.is_empty() {
322            let (next, wait) = self
323                .shared
324                .ready
325                .wait_timeout_while(state, timeout, |state| {
326                    state.runs.is_empty() && !state.interrupted
327                })
328                .map_err(|_| std::sync::mpsc::RecvTimeoutError::Disconnected)?;
329            state = next;
330            let interrupted = std::mem::take(&mut state.interrupted);
331            if state.runs.is_empty() && (interrupted || wait.timed_out()) {
332                return Err(std::sync::mpsc::RecvTimeoutError::Timeout);
333            }
334        }
335        take_direct_source_runs(&mut state, packet_limit)
336            .ok_or(std::sync::mpsc::RecvTimeoutError::Timeout)
337    }
338
339    /// Receive a ready source-contiguous packet run batch without blocking.
340    pub fn try_recv(
341        &self,
342        packet_limit: usize,
343    ) -> Result<Vec<FipsEndpointDirectPacketRun>, std::sync::mpsc::TryRecvError> {
344        let mut state = self
345            .shared
346            .state
347            .lock()
348            .map_err(|_| std::sync::mpsc::TryRecvError::Disconnected)?;
349        let limit = direct_receiver_packet_limit(packet_limit);
350        if state.runs.front().is_some_and(|run| run.len() > limit) {
351            return Err(std::sync::mpsc::TryRecvError::Empty);
352        }
353        take_direct_source_runs(&mut state, limit).ok_or(std::sync::mpsc::TryRecvError::Empty)
354    }
355
356    /// Wake a blocking receive, for example during application shutdown.
357    pub fn interrupt(&self) {
358        if let Ok(mut state) = self.shared.state.lock() {
359            state.interrupted = true;
360        }
361        self.shared.ready.notify_all();
362    }
363}
364
365impl FipsEndpointDirectSink for FipsEndpointDirectReceiverSink {
366    fn deliver_endpoint_packet_batch(
367        &self,
368        batch: FipsEndpointDirectPacketBatch,
369    ) -> Result<(), FipsEndpointDirectDeliveryError> {
370        let runs = batch.into_packet_runs();
371        if runs.is_empty() {
372            return Ok(());
373        }
374        let packets = runs
375            .iter()
376            .map(FipsEndpointDirectPacketRun::len)
377            .sum::<usize>();
378        let mut state = self
379            .shared
380            .state
381            .lock()
382            .map_err(|_| FipsEndpointDirectDeliveryError::Unavailable)?;
383        let queued = state
384            .packets
385            .checked_add(packets)
386            .filter(|queued| *queued <= FIPS_ENDPOINT_DIRECT_PACKET_QUEUE_MAX_PACKETS)
387            .ok_or(FipsEndpointDirectDeliveryError::Unavailable)?;
388        let wake = state.runs.is_empty();
389        state.packets = queued;
390        state.runs.extend(runs);
391        drop(state);
392        if wake {
393            self.shared.ready.notify_one();
394        }
395        Ok(())
396    }
397}
398
399fn direct_receiver_packet_limit(limit: usize) -> usize {
400    limit
401        .max(1)
402        .min(FIPS_ENDPOINT_DIRECT_PACKET_RUN_MAX_PACKETS)
403}
404
405fn take_direct_source_runs(
406    state: &mut FipsEndpointDirectReceiverState,
407    packet_limit: usize,
408) -> Option<Vec<FipsEndpointDirectPacketRun>> {
409    let first = state.runs.pop_front()?;
410    let source = *first.source_peer().node_addr();
411    let mut packets = first.len();
412    let mut runs = vec![first];
413    let limit = direct_receiver_packet_limit(packet_limit);
414    while packets < limit {
415        let Some(next) = state.runs.front() else {
416            break;
417        };
418        let next_packets = next.len();
419        if next.source_peer().node_addr() != &source || packets.saturating_add(next_packets) > limit
420        {
421            break;
422        }
423        packets = packets.saturating_add(next_packets);
424        runs.push(state.runs.pop_front().expect("front direct run must exist"));
425    }
426    state.packets = state
427        .packets
428        .checked_sub(packets)
429        .expect("direct receiver packet backlog must cover removed runs");
430    Some(runs)
431}
432
433/// App-owned packet channels for embedding FIPS without a system TUN.
434#[derive(Debug)]
435pub struct ExternalPacketIo {
436    /// Send outbound IPv6 packets into the node.
437    pub outbound_tx: crate::upper::tun::TunOutboundTx,
438    /// Receive inbound IPv6 packets delivered by FIPS sessions.
439    pub inbound_rx: tokio::sync::mpsc::Receiver<NodeDeliveredPacket>,
440}
441
442/// App-owned endpoint data channels for embedding FIPS without a daemon.
443#[derive(Debug)]
444pub(crate) struct EndpointDataIo {
445    /// Send endpoint management commands into the node RX loop ahead of queued
446    /// endpoint data.
447    pub(crate) control_tx: tokio::sync::mpsc::Sender<NodeEndpointControlCommand>,
448    /// Send endpoint data batches into the node RX loop.
449    ///
450    /// Bounded by the explicit endpoint packet capacity. Bulk backpressure is
451    /// visible to the caller instead of hidden behind an environment-selected
452    /// queue size.
453    pub(crate) data_batch_tx: EndpointDataBatchTx,
454    /// Receive endpoint data delivered by FIPS sessions.
455    ///
456    /// Endpoint data uses one bounded app-data channel. Oversized batches split
457    /// at the message-credit boundary before any remaining tail drops visibly
458    /// via `endpoint_event_bulk_dropped`. Backpressure is still visible through
459    /// `endpoint_event_wait` latency and `endpoint_event_backlog_high` when the
460    /// consumer falls materially behind.
461    pub(crate) event_rx: EndpointEventReceiver,
462    /// Clone of the event_tx exposed for in-process loopback. Lets the endpoint
463    /// inject an event into the same queue without going through the encrypt /
464    /// decrypt path, while keeping every consumer reading from a single channel.
465    pub(crate) event_tx: EndpointEventSender,
466    /// Receive registered FSP service datagrams.
467    pub(crate) service_event_rx: EndpointServiceEventReceiver,
468    /// Clone used for registered in-process loopback service sends.
469    pub(crate) service_event_tx: EndpointServiceEventSender,
470}
471
472/// Observable owner for endpoint events delivered to embedded applications.
473#[derive(Debug, Clone)]
474pub(crate) struct EndpointEventSender {
475    tx: tokio::sync::mpsc::Sender<NodeEndpointEvent>,
476    direct_sink: Option<EndpointDirectSink>,
477    queued_messages: Arc<AtomicUsize>,
478    ready: Arc<EndpointEventReady>,
479    message_cap: usize,
480}
481
482#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
483pub(crate) enum EndpointEventSendError {
484    #[error("endpoint event channel closed")]
485    Closed,
486}
487
488#[derive(Debug)]
489pub(crate) struct EndpointEventReceiver {
490    rx: tokio::sync::mpsc::Receiver<NodeEndpointEvent>,
491    queued_messages: Arc<AtomicUsize>,
492    ready: Arc<EndpointEventReady>,
493    closed: bool,
494}
495
496#[derive(Debug, Default)]
497struct EndpointEventReady {
498    sequence: StdMutex<u64>,
499    changed: Condvar,
500}
501
502impl EndpointEventReady {
503    fn notify(&self) {
504        if let Ok(mut sequence) = self.sequence.lock() {
505            *sequence = sequence.wrapping_add(1);
506            self.changed.notify_one();
507        }
508    }
509
510    fn snapshot(&self) -> u64 {
511        self.sequence.lock().map(|sequence| *sequence).unwrap_or(0)
512    }
513
514    fn wait_for_change(&self, observed: &mut u64) {
515        let Ok(mut sequence) = self.sequence.lock() else {
516            return;
517        };
518        while *sequence == *observed {
519            match self.changed.wait(sequence) {
520                Ok(next) => sequence = next,
521                Err(_) => return,
522            }
523        }
524        *observed = *sequence;
525    }
526}
527
528fn endpoint_event_capacity(requested: usize) -> usize {
529    requested.max(1)
530}
531
532fn try_reserve_endpoint_event_messages(
533    counter: &AtomicUsize,
534    capacity: usize,
535    count: usize,
536) -> Option<usize> {
537    if count == 0 {
538        return Some(counter.load(Relaxed));
539    }
540
541    counter
542        .fetch_update(Relaxed, Relaxed, |current| {
543            current.checked_add(count).filter(|next| *next <= capacity)
544        })
545        .ok()
546}
547
548/// Delivery-side owner for endpoint data emitted by session receive handling.
549///
550/// The rx loop currently owns this runtime, but keeping sender, batching, and
551/// backlog accounting behind one value makes the future peer/shard receive
552/// runtime move explicit instead of threading endpoint-event fields through
553/// `Node` packet handlers.
554#[derive(Debug, Default)]
555pub(in crate::node) struct EndpointEventRuntime {
556    sender: Option<EndpointEventSender>,
557}
558
559impl EndpointEventSender {
560    pub(in crate::node) fn channel(capacity: usize) -> (Self, EndpointEventReceiver) {
561        Self::channel_with_direct_sink(capacity, None)
562    }
563
564    pub(in crate::node) fn channel_with_direct_sink(
565        capacity: usize,
566        direct_sink: Option<EndpointDirectSink>,
567    ) -> (Self, EndpointEventReceiver) {
568        let message_cap = endpoint_event_capacity(capacity);
569        let (tx, rx) = tokio::sync::mpsc::channel(message_cap);
570        let queued_messages = Arc::new(AtomicUsize::new(0));
571        let ready = Arc::new(EndpointEventReady::default());
572        (
573            Self {
574                tx,
575                direct_sink,
576                queued_messages: Arc::clone(&queued_messages),
577                ready: Arc::clone(&ready),
578                message_cap,
579            },
580            EndpointEventReceiver {
581                rx,
582                queued_messages,
583                ready,
584                closed: false,
585            },
586        )
587    }
588
589    pub(crate) fn direct_sink(&self) -> Option<&EndpointDirectSink> {
590        self.direct_sink.as_ref()
591    }
592
593    pub(crate) fn send(&self, event: NodeEndpointEvent) -> Result<(), EndpointEventSendError> {
594        if event.messages.is_empty() {
595            return Ok(());
596        }
597
598        self.send_event(event, true)
599    }
600
601    fn send_event(
602        &self,
603        event: NodeEndpointEvent,
604        split_on_pressure: bool,
605    ) -> Result<(), EndpointEventSendError> {
606        let count = event.message_count();
607        let Some(previous) =
608            try_reserve_endpoint_event_messages(&self.queued_messages, self.message_cap, count)
609        else {
610            if split_on_pressure && count > 1 {
611                return self.split_and_send_event(event);
612            }
613            crate::perf_profile::record_event_count(
614                crate::perf_profile::Event::EndpointEventBulkDropped,
615                count as u64,
616            );
617            return Ok(());
618        };
619
620        let queued = previous.saturating_add(count);
621        match self.tx.try_send(event) {
622            Ok(()) => {
623                self.note_send_success(previous, queued);
624                Ok(())
625            }
626            Err(tokio::sync::mpsc::error::TrySendError::Full(_event)) => {
627                self.note_send_rejected(count);
628                crate::perf_profile::record_event_count(
629                    crate::perf_profile::Event::EndpointEventBulkDropped,
630                    count as u64,
631                );
632                Ok(())
633            }
634            Err(tokio::sync::mpsc::error::TrySendError::Closed(event)) => {
635                self.note_send_rejected(count);
636                drop(event);
637                Err(EndpointEventSendError::Closed)
638            }
639        }
640    }
641
642    fn split_and_send_event(&self, event: NodeEndpointEvent) -> Result<(), EndpointEventSendError> {
643        let mut messages = event.messages;
644        let queued_at = event.queued_at;
645        if messages.len() <= 1 {
646            return self.send_event(
647                NodeEndpointEvent {
648                    messages,
649                    queued_at,
650                },
651                false,
652            );
653        }
654
655        let right = messages.split_off(messages.len() / 2);
656        if !messages.is_empty() {
657            self.send_event(
658                NodeEndpointEvent {
659                    messages,
660                    queued_at,
661                },
662                true,
663            )?;
664        }
665        if !right.is_empty() {
666            self.send_event(
667                NodeEndpointEvent {
668                    messages: right,
669                    queued_at,
670                },
671                true,
672            )?;
673        }
674        Ok(())
675    }
676
677    fn note_send_success(&self, previous: usize, queued: usize) {
678        if previous < ENDPOINT_EVENT_BACKLOG_HIGH_WATER
679            && queued >= ENDPOINT_EVENT_BACKLOG_HIGH_WATER
680        {
681            crate::perf_profile::record_event(crate::perf_profile::Event::EndpointEventBacklogHigh);
682        }
683        self.ready.notify();
684    }
685
686    fn note_send_rejected(&self, count: usize) {
687        release_endpoint_event_messages(&self.queued_messages, count);
688        self.ready.notify();
689    }
690
691    #[cfg(test)]
692    pub(crate) fn queued_messages(&self) -> usize {
693        self.queued_messages.load(Relaxed)
694    }
695}
696
697impl Drop for EndpointEventSender {
698    fn drop(&mut self) {
699        self.ready.notify();
700    }
701}
702
703impl Drop for EndpointEventReceiver {
704    fn drop(&mut self) {
705        self.queued_messages.store(0, Relaxed);
706        self.ready.notify();
707    }
708}
709
710impl EndpointEventRuntime {
711    pub(in crate::node) fn attach(&mut self, sender: EndpointEventSender) {
712        self.sender = Some(sender);
713    }
714
715    pub(in crate::node) fn is_attached(&self) -> bool {
716        self.sender.is_some()
717    }
718
719    pub(in crate::node) fn sender(&self) -> Option<EndpointEventSender> {
720        self.sender.clone()
721    }
722
723    pub(in crate::node) fn deliver_endpoint_data_batch(
724        &mut self,
725        messages: Vec<EndpointDataDelivery>,
726    ) -> Result<(), EndpointEventSendError> {
727        if messages.is_empty() {
728            return Ok(());
729        }
730
731        let Some(sender) = &self.sender else {
732            return Ok(());
733        };
734        let _t_deliver =
735            crate::perf_profile::Timer::start(crate::perf_profile::Stage::EndpointDeliver);
736        sender.send(NodeEndpointEvent {
737            messages,
738            queued_at: crate::perf_profile::stamp(),
739        })
740    }
741}
742
743impl EndpointEventReceiver {
744    pub(crate) async fn recv(&mut self) -> Option<NodeEndpointEvent> {
745        let event = self.rx.recv().await?;
746        self.note_observed(&event);
747        Some(event)
748    }
749
750    pub(crate) fn blocking_recv(&mut self) -> Option<NodeEndpointEvent> {
751        let mut observed = self.ready.snapshot();
752        loop {
753            match self.try_recv() {
754                Ok(event) => return Some(event),
755                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => return None,
756                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
757                    self.ready.wait_for_change(&mut observed);
758                }
759            }
760        }
761    }
762
763    pub(crate) fn try_recv(
764        &mut self,
765    ) -> Result<NodeEndpointEvent, tokio::sync::mpsc::error::TryRecvError> {
766        match self.rx.try_recv() {
767            Ok(event) => {
768                self.note_observed(&event);
769                Ok(event)
770            }
771            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
772                if self.closed {
773                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
774                } else {
775                    Err(tokio::sync::mpsc::error::TryRecvError::Empty)
776                }
777            }
778            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
779                self.closed = true;
780                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
781            }
782        }
783    }
784
785    pub(crate) fn release_messages(&self, count: usize) {
786        release_endpoint_event_messages(&self.queued_messages, count);
787    }
788
789    fn note_observed(&self, event: &NodeEndpointEvent) {
790        event.record_dequeue_wait();
791    }
792}
793
794pub(in crate::node) fn release_endpoint_event_messages(counter: &AtomicUsize, count: usize) {
795    if count == 0 {
796        return;
797    }
798
799    let previous = counter.fetch_sub(count, Relaxed);
800    debug_assert!(
801        previous >= count,
802        "endpoint event queued message accounting underflow"
803    );
804}
805
806/// Reports what changed in response to `UpdatePeers`.
807#[derive(Debug, Clone, Default, PartialEq, Eq)]
808pub(crate) struct UpdatePeersOutcome {
809    pub(crate) added: usize,
810    pub(crate) removed: usize,
811    pub(crate) updated: usize,
812    pub(crate) unchanged: usize,
813}
814
815/// Authenticated endpoint data emitted by the session receive path.
816///
817/// Keeping source identity and payload together makes the delivery-side
818/// ownership boundary explicit for the current rx loop and for a future
819/// peer/session runtime that can move endpoint-data delivery off the bounce path.
820#[derive(Debug, Clone)]
821pub(crate) struct EndpointDataDelivery {
822    pub(crate) source_peer: PeerIdentity,
823    pub(crate) payload: PacketBuffer,
824    pub(crate) enqueued_at_ms: u64,
825}
826
827impl EndpointDataDelivery {
828    pub(crate) fn new(source_peer: PeerIdentity, payload: PacketBuffer) -> Self {
829        Self {
830            source_peer,
831            payload,
832            enqueued_at_ms: crate::time::now_ms(),
833        }
834    }
835}
836
837/// Endpoint data events emitted by the node session receive path.
838#[derive(Debug)]
839pub(crate) struct NodeEndpointEvent {
840    pub(crate) messages: Vec<EndpointDataDelivery>,
841    pub(crate) queued_at: Option<crate::perf_profile::TraceStamp>,
842}
843
844impl NodeEndpointEvent {
845    pub(in crate::node) fn message_count(&self) -> usize {
846        self.messages.len()
847    }
848
849    fn record_dequeue_wait(&self) {
850        let queued_at = self.queued_at;
851        if queued_at.is_none() {
852            return;
853        }
854        crate::perf_profile::record_since_count(
855            crate::perf_profile::Stage::EndpointEventWait,
856            queued_at,
857            self.message_count() as u64,
858        );
859    }
860}
861
862/// Authenticated peer state exposed to embedded endpoint callers.
863#[derive(Debug, Clone, PartialEq, Eq)]
864pub(crate) struct NodeEndpointPeer {
865    pub(crate) npub: String,
866    pub(crate) node_addr: NodeAddr,
867    pub(crate) connected: bool,
868    pub(crate) transport_addr: Option<String>,
869    pub(crate) transport_type: Option<String>,
870    pub(crate) link_id: u64,
871    pub(crate) srtt_ms: Option<u64>,
872    pub(crate) srtt_age_ms: Option<u64>,
873    pub(crate) packets_sent: u64,
874    pub(crate) packets_recv: u64,
875    pub(crate) bytes_sent: u64,
876    pub(crate) bytes_recv: u64,
877    pub(crate) rekey_in_progress: bool,
878    pub(crate) rekey_draining: bool,
879    pub(crate) current_k_bit: Option<bool>,
880    pub(crate) last_outbound_route: Option<String>,
881    pub(crate) direct_probe_pending: bool,
882    pub(crate) direct_probe_after_ms: Option<u64>,
883    pub(crate) direct_probe_retry_count: u32,
884    pub(crate) direct_probe_auto_reconnect: bool,
885    pub(crate) direct_probe_expires_at_ms: Option<u64>,
886    pub(crate) nostr_traversal_consecutive_failures: u32,
887    pub(crate) nostr_traversal_in_cooldown: bool,
888    pub(crate) nostr_traversal_cooldown_until_ms: Option<u64>,
889    pub(crate) nostr_traversal_last_observed_skew_ms: Option<i64>,
890}
891
892/// Live Nostr relay state exposed to embedded endpoint callers.
893#[derive(Debug, Clone, PartialEq, Eq)]
894pub(crate) struct NodeEndpointRelayStatus {
895    pub(crate) url: String,
896    pub(crate) status: String,
897}
fips_core/node/endpoint_event.rs

fips_core/node/
endpoint_event.rs
