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.clamp(1, FIPS_ENDPOINT_DIRECT_PACKET_RUN_MAX_PACKETS)
401}
402
403fn take_direct_source_runs(
404    state: &mut FipsEndpointDirectReceiverState,
405    packet_limit: usize,
406) -> Option<Vec<FipsEndpointDirectPacketRun>> {
407    let first = state.runs.pop_front()?;
408    let source = *first.source_peer().node_addr();
409    let mut packets = first.len();
410    let mut runs = vec![first];
411    let limit = direct_receiver_packet_limit(packet_limit);
412    while packets < limit {
413        let Some(next) = state.runs.front() else {
414            break;
415        };
416        let next_packets = next.len();
417        if next.source_peer().node_addr() != &source || packets.saturating_add(next_packets) > limit
418        {
419            break;
420        }
421        packets = packets.saturating_add(next_packets);
422        runs.push(state.runs.pop_front().expect("front direct run must exist"));
423    }
424    state.packets = state
425        .packets
426        .checked_sub(packets)
427        .expect("direct receiver packet backlog must cover removed runs");
428    Some(runs)
429}
430
431/// App-owned packet channels for embedding FIPS without a system TUN.
432#[derive(Debug)]
433pub struct ExternalPacketIo {
434    /// Send outbound IPv6 packets into the node.
435    pub outbound_tx: crate::upper::tun::TunOutboundTx,
436    /// Receive inbound IPv6 packets delivered by FIPS sessions.
437    pub inbound_rx: tokio::sync::mpsc::Receiver<NodeDeliveredPacket>,
438}
439
440/// App-owned endpoint data channels for embedding FIPS without a daemon.
441#[derive(Debug)]
442pub(crate) struct EndpointDataIo {
443    /// Send endpoint management commands into the node RX loop ahead of queued
444    /// endpoint data.
445    pub(crate) control_tx: tokio::sync::mpsc::Sender<NodeEndpointControlCommand>,
446    /// Send endpoint data batches into the node RX loop.
447    ///
448    /// Bounded by the explicit endpoint packet capacity. Bulk backpressure is
449    /// visible to the caller instead of hidden behind an environment-selected
450    /// queue size.
451    pub(crate) data_batch_tx: EndpointDataBatchTx,
452    /// Receive endpoint data delivered by FIPS sessions.
453    ///
454    /// Endpoint data uses one bounded app-data channel. Oversized batches split
455    /// at the message-credit boundary before any remaining tail drops visibly
456    /// via `endpoint_event_bulk_dropped`. Backpressure is still visible through
457    /// `endpoint_event_wait` latency and `endpoint_event_backlog_high` when the
458    /// consumer falls materially behind.
459    pub(crate) event_rx: EndpointEventReceiver,
460    /// Clone of the event_tx exposed for in-process loopback. Lets the endpoint
461    /// inject an event into the same queue without going through the encrypt /
462    /// decrypt path, while keeping every consumer reading from a single channel.
463    pub(crate) event_tx: EndpointEventSender,
464    /// Receive registered FSP service datagrams.
465    pub(crate) service_event_rx: EndpointServiceEventReceiver,
466    /// Clone used for registered in-process loopback service sends.
467    pub(crate) service_event_tx: EndpointServiceEventSender,
468}
469
470/// Observable owner for endpoint events delivered to embedded applications.
471#[derive(Debug, Clone)]
472pub(crate) struct EndpointEventSender {
473    tx: tokio::sync::mpsc::Sender<NodeEndpointEvent>,
474    direct_sink: Option<EndpointDirectSink>,
475    queued_messages: Arc<AtomicUsize>,
476    ready: Arc<EndpointEventReady>,
477    message_cap: usize,
478}
479
480#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
481pub(crate) enum EndpointEventSendError {
482    #[error("endpoint event channel closed")]
483    Closed,
484}
485
486#[derive(Debug)]
487pub(crate) struct EndpointEventReceiver {
488    rx: tokio::sync::mpsc::Receiver<NodeEndpointEvent>,
489    queued_messages: Arc<AtomicUsize>,
490    ready: Arc<EndpointEventReady>,
491    closed: bool,
492}
493
494#[derive(Debug, Default)]
495struct EndpointEventReady {
496    sequence: StdMutex<u64>,
497    changed: Condvar,
498}
499
500impl EndpointEventReady {
501    fn notify(&self) {
502        if let Ok(mut sequence) = self.sequence.lock() {
503            *sequence = sequence.wrapping_add(1);
504            self.changed.notify_one();
505        }
506    }
507
508    fn snapshot(&self) -> u64 {
509        self.sequence.lock().map(|sequence| *sequence).unwrap_or(0)
510    }
511
512    fn wait_for_change(&self, observed: &mut u64) {
513        let Ok(mut sequence) = self.sequence.lock() else {
514            return;
515        };
516        while *sequence == *observed {
517            match self.changed.wait(sequence) {
518                Ok(next) => sequence = next,
519                Err(_) => return,
520            }
521        }
522        *observed = *sequence;
523    }
524}
525
526fn endpoint_event_capacity(requested: usize) -> usize {
527    requested.max(1)
528}
529
530fn try_reserve_endpoint_event_messages(
531    counter: &AtomicUsize,
532    capacity: usize,
533    count: usize,
534) -> Option<usize> {
535    if count == 0 {
536        return Some(counter.load(Relaxed));
537    }
538
539    counter
540        .fetch_update(Relaxed, Relaxed, |current| {
541            current.checked_add(count).filter(|next| *next <= capacity)
542        })
543        .ok()
544}
545
546/// Delivery-side owner for endpoint data emitted by session receive handling.
547///
548/// The rx loop currently owns this runtime, but keeping sender, batching, and
549/// backlog accounting behind one value makes the future peer/shard receive
550/// runtime move explicit instead of threading endpoint-event fields through
551/// `Node` packet handlers.
552#[derive(Debug, Default)]
553pub(in crate::node) struct EndpointEventRuntime {
554    sender: Option<EndpointEventSender>,
555}
556
557impl EndpointEventSender {
558    pub(in crate::node) fn channel(capacity: usize) -> (Self, EndpointEventReceiver) {
559        Self::channel_with_direct_sink(capacity, None)
560    }
561
562    pub(in crate::node) fn channel_with_direct_sink(
563        capacity: usize,
564        direct_sink: Option<EndpointDirectSink>,
565    ) -> (Self, EndpointEventReceiver) {
566        let message_cap = endpoint_event_capacity(capacity);
567        let (tx, rx) = tokio::sync::mpsc::channel(message_cap);
568        let queued_messages = Arc::new(AtomicUsize::new(0));
569        let ready = Arc::new(EndpointEventReady::default());
570        (
571            Self {
572                tx,
573                direct_sink,
574                queued_messages: Arc::clone(&queued_messages),
575                ready: Arc::clone(&ready),
576                message_cap,
577            },
578            EndpointEventReceiver {
579                rx,
580                queued_messages,
581                ready,
582                closed: false,
583            },
584        )
585    }
586
587    pub(crate) fn direct_sink(&self) -> Option<&EndpointDirectSink> {
588        self.direct_sink.as_ref()
589    }
590
591    pub(crate) fn send(&self, event: NodeEndpointEvent) -> Result<(), EndpointEventSendError> {
592        if event.messages.is_empty() {
593            return Ok(());
594        }
595
596        self.send_event(event, true)
597    }
598
599    fn send_event(
600        &self,
601        event: NodeEndpointEvent,
602        split_on_pressure: bool,
603    ) -> Result<(), EndpointEventSendError> {
604        let count = event.message_count();
605        let Some(previous) =
606            try_reserve_endpoint_event_messages(&self.queued_messages, self.message_cap, count)
607        else {
608            if split_on_pressure && count > 1 {
609                return self.split_and_send_event(event);
610            }
611            crate::perf_profile::record_event_count(
612                crate::perf_profile::Event::EndpointEventBulkDropped,
613                count as u64,
614            );
615            return Ok(());
616        };
617
618        let queued = previous.saturating_add(count);
619        match self.tx.try_send(event) {
620            Ok(()) => {
621                self.note_send_success(previous, queued);
622                Ok(())
623            }
624            Err(tokio::sync::mpsc::error::TrySendError::Full(_event)) => {
625                self.note_send_rejected(count);
626                crate::perf_profile::record_event_count(
627                    crate::perf_profile::Event::EndpointEventBulkDropped,
628                    count as u64,
629                );
630                Ok(())
631            }
632            Err(tokio::sync::mpsc::error::TrySendError::Closed(event)) => {
633                self.note_send_rejected(count);
634                drop(event);
635                Err(EndpointEventSendError::Closed)
636            }
637        }
638    }
639
640    fn split_and_send_event(&self, event: NodeEndpointEvent) -> Result<(), EndpointEventSendError> {
641        let mut messages = event.messages;
642        let queued_at = event.queued_at;
643        if messages.len() <= 1 {
644            return self.send_event(
645                NodeEndpointEvent {
646                    messages,
647                    queued_at,
648                },
649                false,
650            );
651        }
652
653        let right = messages.split_off(messages.len() / 2);
654        if !messages.is_empty() {
655            self.send_event(
656                NodeEndpointEvent {
657                    messages,
658                    queued_at,
659                },
660                true,
661            )?;
662        }
663        if !right.is_empty() {
664            self.send_event(
665                NodeEndpointEvent {
666                    messages: right,
667                    queued_at,
668                },
669                true,
670            )?;
671        }
672        Ok(())
673    }
674
675    fn note_send_success(&self, previous: usize, queued: usize) {
676        if previous < ENDPOINT_EVENT_BACKLOG_HIGH_WATER
677            && queued >= ENDPOINT_EVENT_BACKLOG_HIGH_WATER
678        {
679            crate::perf_profile::record_event(crate::perf_profile::Event::EndpointEventBacklogHigh);
680        }
681        self.ready.notify();
682    }
683
684    fn note_send_rejected(&self, count: usize) {
685        release_endpoint_event_messages(&self.queued_messages, count);
686        self.ready.notify();
687    }
688
689    #[cfg(test)]
690    pub(crate) fn queued_messages(&self) -> usize {
691        self.queued_messages.load(Relaxed)
692    }
693}
694
695impl Drop for EndpointEventSender {
696    fn drop(&mut self) {
697        self.ready.notify();
698    }
699}
700
701impl Drop for EndpointEventReceiver {
702    fn drop(&mut self) {
703        self.queued_messages.store(0, Relaxed);
704        self.ready.notify();
705    }
706}
707
708impl EndpointEventRuntime {
709    pub(in crate::node) fn attach(&mut self, sender: EndpointEventSender) {
710        self.sender = Some(sender);
711    }
712
713    pub(in crate::node) fn is_attached(&self) -> bool {
714        self.sender.is_some()
715    }
716
717    pub(in crate::node) fn sender(&self) -> Option<EndpointEventSender> {
718        self.sender.clone()
719    }
720
721    pub(in crate::node) fn deliver_endpoint_data_batch(
722        &mut self,
723        messages: Vec<EndpointDataDelivery>,
724    ) -> Result<(), EndpointEventSendError> {
725        if messages.is_empty() {
726            return Ok(());
727        }
728
729        let Some(sender) = &self.sender else {
730            return Ok(());
731        };
732        let _t_deliver =
733            crate::perf_profile::Timer::start(crate::perf_profile::Stage::EndpointDeliver);
734        sender.send(NodeEndpointEvent {
735            messages,
736            queued_at: crate::perf_profile::stamp(),
737        })
738    }
739}
740
741impl EndpointEventReceiver {
742    pub(crate) async fn recv(&mut self) -> Option<NodeEndpointEvent> {
743        let event = self.rx.recv().await?;
744        self.note_observed(&event);
745        Some(event)
746    }
747
748    pub(crate) fn blocking_recv(&mut self) -> Option<NodeEndpointEvent> {
749        let mut observed = self.ready.snapshot();
750        loop {
751            match self.try_recv() {
752                Ok(event) => return Some(event),
753                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => return None,
754                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
755                    self.ready.wait_for_change(&mut observed);
756                }
757            }
758        }
759    }
760
761    pub(crate) fn try_recv(
762        &mut self,
763    ) -> Result<NodeEndpointEvent, tokio::sync::mpsc::error::TryRecvError> {
764        match self.rx.try_recv() {
765            Ok(event) => {
766                self.note_observed(&event);
767                Ok(event)
768            }
769            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
770                if self.closed {
771                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
772                } else {
773                    Err(tokio::sync::mpsc::error::TryRecvError::Empty)
774                }
775            }
776            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
777                self.closed = true;
778                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
779            }
780        }
781    }
782
783    pub(crate) fn release_messages(&self, count: usize) {
784        release_endpoint_event_messages(&self.queued_messages, count);
785    }
786
787    fn note_observed(&self, event: &NodeEndpointEvent) {
788        event.record_dequeue_wait();
789    }
790}
791
792pub(in crate::node) fn release_endpoint_event_messages(counter: &AtomicUsize, count: usize) {
793    if count == 0 {
794        return;
795    }
796
797    let previous = counter.fetch_sub(count, Relaxed);
798    debug_assert!(
799        previous >= count,
800        "endpoint event queued message accounting underflow"
801    );
802}
803
804/// Reports what changed in response to `UpdatePeers`.
805#[derive(Debug, Clone, Default, PartialEq, Eq)]
806pub(crate) struct UpdatePeersOutcome {
807    pub(crate) added: usize,
808    pub(crate) removed: usize,
809    pub(crate) updated: usize,
810    pub(crate) unchanged: usize,
811}
812
813/// Authenticated endpoint data emitted by the session receive path.
814///
815/// Keeping source identity and payload together makes the delivery-side
816/// ownership boundary explicit for the current rx loop and for a future
817/// peer/session runtime that can move endpoint-data delivery off the bounce path.
818#[derive(Debug, Clone)]
819pub(crate) struct EndpointDataDelivery {
820    pub(crate) source_peer: PeerIdentity,
821    pub(crate) payload: PacketBuffer,
822    pub(crate) enqueued_at_ms: u64,
823}
824
825impl EndpointDataDelivery {
826    pub(crate) fn new(source_peer: PeerIdentity, payload: PacketBuffer) -> Self {
827        Self {
828            source_peer,
829            payload,
830            enqueued_at_ms: crate::time::now_ms(),
831        }
832    }
833}
834
835/// Endpoint data events emitted by the node session receive path.
836#[derive(Debug)]
837pub(crate) struct NodeEndpointEvent {
838    pub(crate) messages: Vec<EndpointDataDelivery>,
839    pub(crate) queued_at: Option<crate::perf_profile::TraceStamp>,
840}
841
842impl NodeEndpointEvent {
843    pub(in crate::node) fn message_count(&self) -> usize {
844        self.messages.len()
845    }
846
847    fn record_dequeue_wait(&self) {
848        let queued_at = self.queued_at;
849        if queued_at.is_none() {
850            return;
851        }
852        crate::perf_profile::record_since_count(
853            crate::perf_profile::Stage::EndpointEventWait,
854            queued_at,
855            self.message_count() as u64,
856        );
857    }
858}
859
860/// Authenticated peer state exposed to embedded endpoint callers.
861#[derive(Debug, Clone, PartialEq, Eq)]
862pub(crate) struct NodeEndpointPeer {
863    pub(crate) npub: String,
864    pub(crate) node_addr: NodeAddr,
865    pub(crate) connected: bool,
866    pub(crate) transport_addr: Option<String>,
867    pub(crate) transport_type: Option<String>,
868    pub(crate) link_id: u64,
869    pub(crate) srtt_ms: Option<u64>,
870    pub(crate) srtt_age_ms: Option<u64>,
871    pub(crate) packets_sent: u64,
872    pub(crate) packets_recv: u64,
873    pub(crate) bytes_sent: u64,
874    pub(crate) bytes_recv: u64,
875    pub(crate) rekey_in_progress: bool,
876    pub(crate) rekey_draining: bool,
877    pub(crate) current_k_bit: Option<bool>,
878    pub(crate) last_outbound_route: Option<String>,
879    pub(crate) direct_probe_pending: bool,
880    pub(crate) direct_probe_after_ms: Option<u64>,
881    pub(crate) direct_probe_retry_count: u32,
882    pub(crate) direct_probe_auto_reconnect: bool,
883    pub(crate) direct_probe_expires_at_ms: Option<u64>,
884    pub(crate) nostr_traversal_consecutive_failures: u32,
885    pub(crate) nostr_traversal_in_cooldown: bool,
886    pub(crate) nostr_traversal_cooldown_until_ms: Option<u64>,
887    pub(crate) nostr_traversal_last_observed_skew_ms: Option<i64>,
888}
889
890/// Live Nostr relay state exposed to embedded endpoint callers.
891#[derive(Debug, Clone, PartialEq, Eq)]
892pub(crate) struct NodeEndpointRelayStatus {
893    pub(crate) url: String,
894    pub(crate) status: String,
895}
fips_core/node/endpoint_event.rs

fips_core/node/
endpoint_event.rs
