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

fips_core/node/
endpoint_event.rs
