fips_core/node/
endpoint_event.rs

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

fips_core/node/
endpoint_event.rs
