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 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    /// Authenticated FIPS peer that originated every packet in this run.
34    pub fn source_peer(&self) -> &PeerIdentity {
35        &self.source_peer
36    }
37
38    /// FIPS node address that originated every packet in this run.
39    pub fn source_node_addr(&self) -> &NodeAddr {
40        self.source_peer.node_addr()
41    }
42
43    /// Source Nostr public key as human-facing bech32 text.
44    pub fn source_npub(&self) -> String {
45        self.source_peer.npub()
46    }
47
48    /// Authenticated previous hop for this established FSP receive run.
49    pub fn previous_hop_node_addr(&self) -> &NodeAddr {
50        &self.previous_hop_addr
51    }
52
53    /// Whether FIPS received the run directly from the source node.
54    pub fn is_direct_path(&self) -> bool {
55        self.direct_path
56    }
57
58    /// Whether the established FSP packet carried the key-epoch bit.
59    pub fn received_k_bit(&self) -> bool {
60        self.received_k_bit
61    }
62
63    /// Unix-millisecond time when FIPS handed this run to the direct sink.
64    pub fn enqueued_at_ms(&self) -> u64 {
65        self.enqueued_at_ms
66    }
67}
68
69/// Consecutive direct endpoint packets from one authenticated FIPS source.
70#[derive(Debug, Clone, PartialEq, Eq)]
71pub struct FipsEndpointDirectSourceRun {
72    source_peer: PeerIdentity,
73    packets: Vec<PacketBuffer>,
74    enqueued_at_ms: u64,
75}
76
77impl FipsEndpointDirectSourceRun {
78    pub(crate) fn from_source_packets(
79        source_peer: PeerIdentity,
80        packets: Vec<PacketBuffer>,
81        enqueued_at_ms: u64,
82    ) -> Self {
83        Self {
84            source_peer,
85            packets,
86            enqueued_at_ms,
87        }
88    }
89
90    /// Authenticated FIPS peer that originated every packet in this run.
91    pub fn source_peer(&self) -> &PeerIdentity {
92        &self.source_peer
93    }
94
95    /// FIPS node address that originated every packet in this run.
96    pub fn source_node_addr(&self) -> &NodeAddr {
97        self.source_peer.node_addr()
98    }
99
100    /// Source Nostr public key as human-facing bech32 text.
101    pub fn source_npub(&self) -> String {
102        self.source_peer.npub()
103    }
104
105    /// Unix-millisecond time when FIPS handed this run to the direct sink.
106    pub fn enqueued_at_ms(&self) -> u64 {
107        self.enqueued_at_ms
108    }
109
110    /// Packets delivered for this source run.
111    pub fn packets(&self) -> &[PacketBuffer] {
112        &self.packets
113    }
114
115    /// Take ownership of the run source and packets.
116    pub fn into_parts(self) -> (PeerIdentity, Vec<PacketBuffer>) {
117        (self.source_peer, self.packets)
118    }
119
120    /// Take ownership of the delivered packets.
121    pub fn into_packets(self) -> Vec<PacketBuffer> {
122        self.packets
123    }
124
125    /// Number of endpoint packets in the run.
126    pub fn len(&self) -> usize {
127        self.packets.len()
128    }
129
130    /// Whether the run contains no packets.
131    pub fn is_empty(&self) -> bool {
132        self.packets.is_empty()
133    }
134}
135
136/// Consecutive direct endpoint packets from one authenticated FIPS source.
137#[derive(Debug, Clone, PartialEq, Eq)]
138struct FipsEndpointDirectPacketSegment {
139    buffer: Arc<PacketBuffer>,
140    ranges: Vec<Range<usize>>,
141    packet_bytes: usize,
142}
143
144impl FipsEndpointDirectPacketSegment {
145    fn new(buffer: PacketBuffer, ranges: Vec<Range<usize>>) -> Self {
146        Self::from_shared_buffer(Arc::new(buffer), ranges)
147    }
148
149    fn from_shared_buffer(buffer: Arc<PacketBuffer>, ranges: Vec<Range<usize>>) -> Self {
150        debug_assert!(ranges.windows(2).all(|pair| pair[0].end <= pair[1].start));
151        let packet_bytes = ranges.iter().map(|range| range.len()).sum();
152        Self {
153            buffer,
154            ranges,
155            packet_bytes,
156        }
157    }
158
159    fn len(&self) -> usize {
160        self.ranges.len()
161    }
162
163    fn is_empty(&self) -> bool {
164        self.ranges.is_empty()
165    }
166
167    fn push_range_from_shared_buffer(
168        &mut self,
169        buffer: &Arc<PacketBuffer>,
170        range: Range<usize>,
171    ) -> bool {
172        if !Arc::ptr_eq(&self.buffer, buffer) {
173            return false;
174        }
175        if self
176            .ranges
177            .last()
178            .is_some_and(|previous| previous.end > range.start)
179        {
180            return false;
181        }
182        self.packet_bytes = self.packet_bytes.saturating_add(range.len());
183        self.ranges.push(range);
184        true
185    }
186}
187
188#[derive(Debug)]
189struct FipsEndpointDirectPacketSplitGroup {
190    lane: usize,
191    segments: Vec<FipsEndpointDirectPacketSegment>,
192}
193
194impl FipsEndpointDirectPacketSplitGroup {
195    fn new(lane: usize) -> Self {
196        Self {
197            lane,
198            segments: Vec::new(),
199        }
200    }
201
202    fn push(&mut self, buffer: Arc<PacketBuffer>, range: Range<usize>) {
203        if let Some(last) = self.segments.last_mut()
204            && last.push_range_from_shared_buffer(&buffer, range.clone())
205        {
206            return;
207        }
208        self.segments
209            .push(FipsEndpointDirectPacketSegment::from_shared_buffer(
210                buffer,
211                vec![range],
212            ));
213    }
214}
215
216/// Consecutive direct endpoint packets from one authenticated FIPS source.
217#[derive(Debug, Clone, PartialEq, Eq)]
218enum FipsEndpointDirectPacketStorage {
219    Segmented(FipsEndpointDirectPacketSegment),
220    Chained {
221        segments: Vec<FipsEndpointDirectPacketSegment>,
222        packet_ends: Vec<usize>,
223        packet_bytes: usize,
224    },
225}
226
227impl FipsEndpointDirectPacketStorage {
228    fn empty_segmented() -> Self {
229        Self::Segmented(FipsEndpointDirectPacketSegment::new(
230            PacketBuffer::new(Vec::new()),
231            Vec::new(),
232        ))
233    }
234
235    fn build_chained(mut segments: Vec<FipsEndpointDirectPacketSegment>) -> Self {
236        let mut packet_ends = Vec::with_capacity(segments.len());
237        let mut packet_count = 0usize;
238        let mut packet_bytes = 0usize;
239        segments.retain(|segment| {
240            if segment.is_empty() {
241                return false;
242            }
243            packet_count = packet_count.saturating_add(segment.len());
244            packet_ends.push(packet_count);
245            packet_bytes = packet_bytes.saturating_add(segment.packet_bytes);
246            true
247        });
248        Self::Chained {
249            segments,
250            packet_ends,
251            packet_bytes,
252        }
253    }
254
255    fn packet_count(&self) -> usize {
256        match self {
257            Self::Segmented(segment) => segment.len(),
258            Self::Chained { packet_ends, .. } => packet_ends.last().copied().unwrap_or(0),
259        }
260    }
261
262    fn into_segments(self) -> Vec<FipsEndpointDirectPacketSegment> {
263        match self {
264            Self::Segmented(segment) => vec![segment],
265            Self::Chained { segments, .. } => segments,
266        }
267    }
268}
269
270/// Consecutive direct endpoint packets from one authenticated FIPS source.
271///
272/// Unlike [`FipsEndpointDirectSourceRun`], this can preserve an opened
273/// EndpointDataBulk buffer and expose packet slices by range. That is the
274/// canonical direct PM2 endpoint payload contract for high-throughput embedders:
275/// FIPS owns authentication and ordering, while the embedder can still apply
276/// live routing policy before borrowing packet bytes for TUN writes.
277#[derive(Debug, Clone, PartialEq, Eq)]
278pub struct FipsEndpointDirectPacketRun {
279    meta: FipsEndpointDirectPacketRunMeta,
280    storage: FipsEndpointDirectPacketStorage,
281}
282
283/// Borrowed packet slices from a direct endpoint packet run.
284pub struct FipsEndpointDirectPacketSlices<'a> {
285    storage: &'a FipsEndpointDirectPacketStorage,
286    index: usize,
287    segment_index: usize,
288    segment_packet_index: usize,
289    remaining: usize,
290}
291
292impl FipsEndpointDirectPacketRun {
293    pub(crate) fn from_segmented_payload(
294        meta: FipsEndpointDirectPacketRunMeta,
295        buffer: PacketBuffer,
296        ranges: Vec<Range<usize>>,
297    ) -> Self {
298        Self {
299            meta,
300            storage: FipsEndpointDirectPacketStorage::Segmented(
301                FipsEndpointDirectPacketSegment::new(buffer, ranges),
302            ),
303        }
304    }
305
306    /// Authenticated source/session facts for this packet run.
307    pub fn meta(&self) -> &FipsEndpointDirectPacketRunMeta {
308        &self.meta
309    }
310
311    /// Authenticated FIPS peer that originated every packet in this run.
312    pub fn source_peer(&self) -> &PeerIdentity {
313        self.meta.source_peer()
314    }
315
316    /// FIPS node address that originated every packet in this run.
317    pub fn source_node_addr(&self) -> &NodeAddr {
318        self.meta.source_node_addr()
319    }
320
321    /// Source Nostr public key as human-facing bech32 text.
322    pub fn source_npub(&self) -> String {
323        self.meta.source_npub()
324    }
325
326    /// Authenticated previous hop for this established FSP receive run.
327    pub fn previous_hop_node_addr(&self) -> &NodeAddr {
328        self.meta.previous_hop_node_addr()
329    }
330
331    /// Whether FIPS received the run directly from the source node.
332    pub fn is_direct_path(&self) -> bool {
333        self.meta.is_direct_path()
334    }
335
336    /// Whether the established FSP packet carried the key-epoch bit.
337    pub fn received_k_bit(&self) -> bool {
338        self.meta.received_k_bit()
339    }
340
341    /// Unix-millisecond time when FIPS handed this run to the direct sink.
342    pub fn enqueued_at_ms(&self) -> u64 {
343        self.meta.enqueued_at_ms()
344    }
345
346    /// Number of endpoint packets in the run.
347    pub fn len(&self) -> usize {
348        self.storage.packet_count()
349    }
350
351    /// Whether the run contains no packets.
352    pub fn is_empty(&self) -> bool {
353        self.len() == 0
354    }
355
356    /// Sum of endpoint packet bytes, excluding bulk length metadata.
357    pub fn packet_bytes(&self) -> usize {
358        match &self.storage {
359            FipsEndpointDirectPacketStorage::Segmented(segment) => segment.packet_bytes,
360            FipsEndpointDirectPacketStorage::Chained { packet_bytes, .. } => *packet_bytes,
361        }
362    }
363
364    /// Borrow one packet by index.
365    pub fn packet_slice(&self, index: usize) -> Option<&[u8]> {
366        match &self.storage {
367            FipsEndpointDirectPacketStorage::Segmented(segment) => segment
368                .ranges
369                .get(index)
370                .map(|range| &segment.buffer.as_slice()[range.clone()]),
371            FipsEndpointDirectPacketStorage::Chained {
372                segments,
373                packet_ends,
374                ..
375            } => {
376                let segment_index = packet_ends.partition_point(|end| *end <= index);
377                let previous_end = segment_index
378                    .checked_sub(1)
379                    .and_then(|previous| packet_ends.get(previous).copied())
380                    .unwrap_or(0);
381                segments.get(segment_index).and_then(|segment| {
382                    segment
383                        .ranges
384                        .get(index - previous_end)
385                        .map(|range| &segment.buffer.as_slice()[range.clone()])
386                })
387            }
388        }
389    }
390
391    /// Mutably borrow one packet by index.
392    pub fn packet_slice_mut(&mut self, index: usize) -> Option<&mut [u8]> {
393        match &mut self.storage {
394            FipsEndpointDirectPacketStorage::Segmented(segment) => {
395                let range = segment.ranges.get(index)?.clone();
396                Some(&mut Arc::make_mut(&mut segment.buffer).as_mut_slice()[range])
397            }
398            FipsEndpointDirectPacketStorage::Chained {
399                segments,
400                packet_ends,
401                ..
402            } => {
403                let segment_index = packet_ends.partition_point(|end| *end <= index);
404                let previous_end = segment_index
405                    .checked_sub(1)
406                    .and_then(|previous| packet_ends.get(previous).copied())
407                    .unwrap_or(0);
408                let segment = segments.get_mut(segment_index)?;
409                let range = segment.ranges.get(index - previous_end)?.clone();
410                Some(&mut Arc::make_mut(&mut segment.buffer).as_mut_slice()[range])
411            }
412        }
413    }
414
415    pub(crate) fn try_append_run(
416        &mut self,
417        other: FipsEndpointDirectPacketRun,
418    ) -> Result<(), FipsEndpointDirectPacketRun> {
419        if !self.matches_append_meta(&other) {
420            return Err(other);
421        }
422
423        let current = std::mem::replace(
424            &mut self.storage,
425            FipsEndpointDirectPacketStorage::empty_segmented(),
426        );
427        let mut segments = match current {
428            FipsEndpointDirectPacketStorage::Segmented(segment) => vec![segment],
429            FipsEndpointDirectPacketStorage::Chained { segments, .. } => segments,
430        };
431        match other.storage {
432            FipsEndpointDirectPacketStorage::Segmented(segment) => segments.push(segment),
433            FipsEndpointDirectPacketStorage::Chained {
434                segments: mut other_segments,
435                ..
436            } => segments.append(&mut other_segments),
437        }
438        self.storage = FipsEndpointDirectPacketStorage::build_chained(segments);
439        Ok(())
440    }
441
442    fn matches_append_meta(&self, other: &Self) -> bool {
443        self.source_peer() == other.source_peer()
444            && self.previous_hop_node_addr() == other.previous_hop_node_addr()
445            && self.received_k_bit() == other.received_k_bit()
446            && self.is_direct_path() == other.is_direct_path()
447    }
448
449    /// Borrow packet bytes without materializing per-packet buffers.
450    pub fn packet_slices(&self) -> FipsEndpointDirectPacketSlices<'_> {
451        FipsEndpointDirectPacketSlices {
452            storage: &self.storage,
453            index: 0,
454            segment_index: 0,
455            segment_packet_index: 0,
456            remaining: self.len(),
457        }
458    }
459
460    /// Partition this run into packet-lane groups without copying packet bytes.
461    ///
462    /// The caller chooses a lane from immutable endpoint packet bytes. FIPS keeps
463    /// authentication/session metadata on every child run and shares the opened
464    /// endpoint payload buffer across lane runs.
465    pub fn partition_by_packet_lane<F>(
466        self,
467        lane_count: usize,
468        mut lane_for_packet: F,
469    ) -> Vec<(usize, Self)>
470    where
471        F: FnMut(&[u8]) -> usize,
472    {
473        let meta = self.meta;
474        let mut groups: Vec<FipsEndpointDirectPacketSplitGroup> = Vec::new();
475        for segment in self.storage.into_segments() {
476            let buffer = segment.buffer;
477            let bytes = buffer.as_slice();
478            for range in segment.ranges {
479                let lane = if lane_count == 0 {
480                    0
481                } else {
482                    lane_for_packet(&bytes[range.clone()]) % lane_count
483                };
484                let group_index = groups.iter().position(|group| group.lane == lane);
485                let group = match group_index {
486                    Some(index) => &mut groups[index],
487                    None => {
488                        groups.push(FipsEndpointDirectPacketSplitGroup::new(lane));
489                        groups.last_mut().expect("group was just pushed")
490                    }
491                };
492                group.push(Arc::clone(&buffer), range);
493            }
494        }
495
496        groups
497            .into_iter()
498            .map(|group| {
499                let run = Self {
500                    meta: meta.clone(),
501                    storage: FipsEndpointDirectPacketStorage::build_chained(group.segments),
502                };
503                (group.lane, run)
504            })
505            .collect()
506    }
507
508    /// Keep only packets accepted by the caller while preserving backing storage.
509    ///
510    /// The predicate receives the original packet index and immutable bytes. This
511    /// keeps routing/admission policy outside FIPS while allowing embedders to
512    /// remove rejected ranges before a TUN writer borrows or mutates the run.
513    pub fn retain_packets<F>(&mut self, mut keep: F)
514    where
515        F: FnMut(usize, &[u8]) -> bool,
516    {
517        match &mut self.storage {
518            FipsEndpointDirectPacketStorage::Segmented(segment) => {
519                let bytes = segment.buffer.as_slice();
520                let mut index = 0usize;
521                let mut retained_bytes = 0usize;
522                segment.ranges.retain(|range| {
523                    let current_index = index;
524                    index = index.saturating_add(1);
525                    if keep(current_index, &bytes[range.clone()]) {
526                        retained_bytes = retained_bytes.saturating_add(range.len());
527                        true
528                    } else {
529                        false
530                    }
531                });
532                segment.packet_bytes = retained_bytes;
533            }
534            FipsEndpointDirectPacketStorage::Chained {
535                segments,
536                packet_ends,
537                packet_bytes,
538            } => {
539                let mut index = 0usize;
540                let mut retained_bytes = 0usize;
541                for segment in segments.iter_mut() {
542                    let bytes = segment.buffer.as_slice();
543                    let mut segment_retained_bytes = 0usize;
544                    segment.ranges.retain(|range| {
545                        let current_index = index;
546                        index = index.saturating_add(1);
547                        if keep(current_index, &bytes[range.clone()]) {
548                            retained_bytes = retained_bytes.saturating_add(range.len());
549                            segment_retained_bytes =
550                                segment_retained_bytes.saturating_add(range.len());
551                            true
552                        } else {
553                            false
554                        }
555                    });
556                    segment.packet_bytes = segment_retained_bytes;
557                }
558                segments.retain(|segment| !segment.is_empty());
559                packet_ends.clear();
560                let mut packet_count = 0usize;
561                for segment in segments.iter() {
562                    packet_count = packet_count.saturating_add(segment.len());
563                    packet_ends.push(packet_count);
564                }
565                *packet_bytes = retained_bytes;
566            }
567        }
568    }
569
570    /// Visit each packet as mutable bytes while the run owner is borrowed.
571    pub fn for_each_packet_mut<F>(&mut self, mut visit: F)
572    where
573        F: FnMut(&mut [u8]),
574    {
575        match &mut self.storage {
576            FipsEndpointDirectPacketStorage::Segmented(segment) => {
577                let bytes = Arc::make_mut(&mut segment.buffer).as_mut_slice();
578                for range in &segment.ranges {
579                    visit(&mut bytes[range.clone()]);
580                }
581            }
582            FipsEndpointDirectPacketStorage::Chained { segments, .. } => {
583                for segment in segments {
584                    let bytes = Arc::make_mut(&mut segment.buffer).as_mut_slice();
585                    for range in &segment.ranges {
586                        visit(&mut bytes[range.clone()]);
587                    }
588                }
589            }
590        }
591    }
592
593    /// Materialize this run into the older owned-packet source-run contract.
594    pub fn into_source_run(self) -> FipsEndpointDirectSourceRun {
595        match self.storage {
596            FipsEndpointDirectPacketStorage::Segmented(segment) => {
597                let body = segment.buffer.as_slice();
598                let packets = segment
599                    .ranges
600                    .into_iter()
601                    .map(|range| body[range].to_vec().into())
602                    .collect();
603                FipsEndpointDirectSourceRun::from_source_packets(
604                    self.meta.source_peer,
605                    packets,
606                    self.meta.enqueued_at_ms,
607                )
608            }
609            FipsEndpointDirectPacketStorage::Chained { segments, .. } => {
610                let mut packets = Vec::new();
611                for segment in segments {
612                    let body = segment.buffer.as_slice();
613                    packets.extend(
614                        segment
615                            .ranges
616                            .into_iter()
617                            .map(|range| body[range].to_vec().into()),
618                    );
619                }
620                FipsEndpointDirectSourceRun::from_source_packets(
621                    self.meta.source_peer,
622                    packets,
623                    self.meta.enqueued_at_ms,
624                )
625            }
626        }
627    }
628
629    /// Materialize this run into owned packet buffers.
630    pub fn into_packets(self) -> Vec<PacketBuffer> {
631        self.into_source_run().into_packets()
632    }
633}
634
635impl<'a> Iterator for FipsEndpointDirectPacketSlices<'a> {
636    type Item = &'a [u8];
637
638    fn next(&mut self) -> Option<Self::Item> {
639        if self.remaining == 0 {
640            return None;
641        }
642        let packet = match self.storage {
643            FipsEndpointDirectPacketStorage::Segmented(segment) => segment
644                .ranges
645                .get(self.index)
646                .map(|range| &segment.buffer.as_slice()[range.clone()]),
647            FipsEndpointDirectPacketStorage::Chained { segments, .. } => loop {
648                let Some(segment) = segments.get(self.segment_index) else {
649                    break None;
650                };
651                if self.segment_packet_index < segment.len() {
652                    let packet = segment
653                        .ranges
654                        .get(self.segment_packet_index)
655                        .map(|range| &segment.buffer.as_slice()[range.clone()]);
656                    self.segment_packet_index = self.segment_packet_index.saturating_add(1);
657                    if self.segment_packet_index >= segment.len() {
658                        self.segment_index = self.segment_index.saturating_add(1);
659                        self.segment_packet_index = 0;
660                    }
661                    break packet;
662                }
663                self.segment_index = self.segment_index.saturating_add(1);
664                self.segment_packet_index = 0;
665            },
666        };
667        if packet.is_some() {
668            self.index = self.index.saturating_add(1);
669            self.remaining = self.remaining.saturating_sub(1);
670        }
671        packet
672    }
673
674    fn size_hint(&self) -> (usize, Option<usize>) {
675        (self.remaining, Some(self.remaining))
676    }
677}
678
679impl ExactSizeIterator for FipsEndpointDirectPacketSlices<'_> {}
680
681/// Established endpoint packet runs delivered without the endpoint-event queue.
682#[derive(Debug, Clone, PartialEq, Eq)]
683pub struct FipsEndpointDirectPacketBatch {
684    packet_runs: Vec<FipsEndpointDirectPacketRun>,
685}
686
687impl FipsEndpointDirectPacketBatch {
688    pub(crate) fn from_packet_runs(packet_runs: Vec<FipsEndpointDirectPacketRun>) -> Self {
689        Self { packet_runs }
690    }
691
692    /// Packet runs in this direct delivery batch.
693    pub fn packet_runs(&self) -> &[FipsEndpointDirectPacketRun] {
694        &self.packet_runs
695    }
696
697    /// Mutably borrow packet runs so the embedder can apply live policy.
698    pub fn packet_runs_mut(&mut self) -> &mut [FipsEndpointDirectPacketRun] {
699        &mut self.packet_runs
700    }
701
702    /// Take ownership of the delivered packet runs.
703    pub fn into_packet_runs(self) -> Vec<FipsEndpointDirectPacketRun> {
704        self.packet_runs
705    }
706
707    /// Whether every run in this batch came from the same FIPS node.
708    pub fn is_single_source(&self) -> bool {
709        self.packet_runs
710            .windows(2)
711            .all(|pair| pair[0].source_node_addr() == pair[1].source_node_addr())
712    }
713
714    /// Number of endpoint messages in the batch.
715    pub fn len(&self) -> usize {
716        self.packet_runs
717            .iter()
718            .map(FipsEndpointDirectPacketRun::len)
719            .sum()
720    }
721
722    /// Sum of endpoint packet bytes in the batch.
723    pub fn packet_bytes(&self) -> usize {
724        self.packet_runs
725            .iter()
726            .map(FipsEndpointDirectPacketRun::packet_bytes)
727            .sum()
728    }
729
730    /// Number of packet-run records in the batch.
731    pub fn run_count(&self) -> usize {
732        self.packet_runs.len()
733    }
734
735    /// Whether the batch contains no packet runs.
736    pub fn is_empty(&self) -> bool {
737        self.packet_runs.is_empty()
738    }
739}
740
741/// Error returned by an installed direct endpoint sink.
742#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
743pub enum FipsEndpointDirectDeliveryError {
744    /// The sink could not accept this batch.
745    #[error("direct endpoint sink unavailable")]
746    Unavailable,
747}
748
749/// Application-provided direct PM2 endpoint delivery sink.
750///
751/// This sink is called synchronously from the PM2 output path with owned packet
752/// buffers. It should return quickly and avoid blocking unrelated PM2 progress.
753pub trait FipsEndpointDirectSink: Send + Sync + 'static {
754    /// Deliver established endpoint data as authenticated packet runs.
755    fn deliver_endpoint_packet_batch(
756        &self,
757        batch: FipsEndpointDirectPacketBatch,
758    ) -> Result<(), FipsEndpointDirectDeliveryError>;
759}
760
761impl<F> FipsEndpointDirectSink for F
762where
763    F: Fn(FipsEndpointDirectPacketBatch) -> Result<(), FipsEndpointDirectDeliveryError>
764        + Send
765        + Sync
766        + 'static,
767{
768    fn deliver_endpoint_packet_batch(
769        &self,
770        batch: FipsEndpointDirectPacketBatch,
771    ) -> Result<(), FipsEndpointDirectDeliveryError> {
772        self(batch)
773    }
774}
775
776#[derive(Clone)]
777pub(crate) struct EndpointDirectSink {
778    sink: Arc<dyn FipsEndpointDirectSink>,
779}
780
781impl std::fmt::Debug for EndpointDirectSink {
782    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
783        f.debug_struct("EndpointDirectSink").finish_non_exhaustive()
784    }
785}
786
787impl EndpointDirectSink {
788    pub(crate) fn new<S>(sink: S) -> Self
789    where
790        S: FipsEndpointDirectSink,
791    {
792        Self {
793            sink: Arc::new(sink),
794        }
795    }
796
797    pub(crate) fn deliver_direct_packet_batch(
798        &self,
799        batch: FipsEndpointDirectPacketBatch,
800    ) -> Result<(), FipsEndpointDirectDeliveryError> {
801        self.sink.deliver_endpoint_packet_batch(batch)
802    }
803}
804
805/// App-owned packet channels for embedding FIPS without a system TUN.
806#[derive(Debug)]
807pub struct ExternalPacketIo {
808    /// Send outbound IPv6 packets into the node.
809    pub outbound_tx: crate::upper::tun::TunOutboundTx,
810    /// Receive inbound IPv6 packets delivered by FIPS sessions.
811    pub inbound_rx: tokio::sync::mpsc::Receiver<NodeDeliveredPacket>,
812}
813
814/// App-owned endpoint data channels for embedding FIPS without a daemon.
815#[derive(Debug)]
816pub(crate) struct EndpointDataIo {
817    /// Send endpoint management commands into the node RX loop ahead of queued
818    /// endpoint data.
819    pub(crate) control_tx: tokio::sync::mpsc::Sender<NodeEndpointControlCommand>,
820    /// Send endpoint data batches into the node RX loop.
821    ///
822    /// Bounded by the explicit endpoint packet capacity. Bulk backpressure is
823    /// visible to the caller instead of hidden behind an environment-selected
824    /// queue size.
825    pub(crate) data_batch_tx: EndpointDataBatchTx,
826    /// Receive endpoint data delivered by FIPS sessions.
827    ///
828    /// Endpoint data uses one bounded app-data channel. Oversized batches split
829    /// at the message-credit boundary before any remaining tail drops visibly
830    /// via `endpoint_event_bulk_dropped`. Backpressure is still visible through
831    /// `endpoint_event_wait` latency and `endpoint_event_backlog_high` when the
832    /// consumer falls materially behind.
833    pub(crate) event_rx: EndpointEventReceiver,
834    /// Clone of the event_tx exposed for in-process loopback (e.g.
835    /// `FipsEndpoint::send` to self_npub). Lets the endpoint inject an
836    /// event into the same queue without going through the encrypt /
837    /// decrypt path, while keeping every consumer reading from a single
838    /// channel.
839    pub(crate) event_tx: EndpointEventSender,
840}
841
842/// Observable owner for endpoint events delivered to embedded applications.
843#[derive(Debug, Clone)]
844pub(crate) struct EndpointEventSender {
845    tx: tokio::sync::mpsc::Sender<NodeEndpointEvent>,
846    direct_sink: Option<EndpointDirectSink>,
847    queued_messages: Arc<AtomicUsize>,
848    ready: Arc<EndpointEventReady>,
849    message_cap: usize,
850}
851
852#[derive(Debug)]
853pub(crate) struct EndpointEventReceiver {
854    rx: tokio::sync::mpsc::Receiver<NodeEndpointEvent>,
855    queued_messages: Arc<AtomicUsize>,
856    ready: Arc<EndpointEventReady>,
857    closed: bool,
858}
859
860#[derive(Debug, Default)]
861struct EndpointEventReady {
862    sequence: StdMutex<u64>,
863    changed: Condvar,
864}
865
866impl EndpointEventReady {
867    fn notify(&self) {
868        if let Ok(mut sequence) = self.sequence.lock() {
869            *sequence = sequence.wrapping_add(1);
870            self.changed.notify_one();
871        }
872    }
873
874    fn snapshot(&self) -> u64 {
875        self.sequence.lock().map(|sequence| *sequence).unwrap_or(0)
876    }
877
878    fn wait_for_change(&self, observed: &mut u64) {
879        let Ok(mut sequence) = self.sequence.lock() else {
880            return;
881        };
882        while *sequence == *observed {
883            match self.changed.wait(sequence) {
884                Ok(next) => sequence = next,
885                Err(_) => return,
886            }
887        }
888        *observed = *sequence;
889    }
890}
891
892fn endpoint_event_capacity(requested: usize) -> usize {
893    requested.max(1)
894}
895
896fn try_reserve_endpoint_event_messages(
897    counter: &AtomicUsize,
898    capacity: usize,
899    count: usize,
900) -> Option<usize> {
901    if count == 0 {
902        return Some(counter.load(Relaxed));
903    }
904
905    counter
906        .fetch_update(Relaxed, Relaxed, |current| {
907            current.checked_add(count).filter(|next| *next <= capacity)
908        })
909        .ok()
910}
911
912/// Delivery-side owner for endpoint data emitted by session receive handling.
913///
914/// The rx loop currently owns this runtime, but keeping sender, batching, and
915/// backlog accounting behind one value makes the future peer/shard receive
916/// runtime move explicit instead of threading endpoint-event fields through
917/// `Node` packet handlers.
918#[derive(Debug, Default)]
919pub(in crate::node) struct EndpointEventRuntime {
920    sender: Option<EndpointEventSender>,
921}
922
923impl EndpointEventSender {
924    pub(in crate::node) fn channel(capacity: usize) -> (Self, EndpointEventReceiver) {
925        Self::channel_with_direct_sink(capacity, None)
926    }
927
928    pub(in crate::node) fn channel_with_direct_sink(
929        capacity: usize,
930        direct_sink: Option<EndpointDirectSink>,
931    ) -> (Self, EndpointEventReceiver) {
932        let message_cap = endpoint_event_capacity(capacity);
933        let (tx, rx) = tokio::sync::mpsc::channel(message_cap);
934        let queued_messages = Arc::new(AtomicUsize::new(0));
935        let ready = Arc::new(EndpointEventReady::default());
936        (
937            Self {
938                tx,
939                direct_sink,
940                queued_messages: Arc::clone(&queued_messages),
941                ready: Arc::clone(&ready),
942                message_cap,
943            },
944            EndpointEventReceiver {
945                rx,
946                queued_messages,
947                ready,
948                closed: false,
949            },
950        )
951    }
952
953    pub(crate) fn direct_sink(&self) -> Option<&EndpointDirectSink> {
954        self.direct_sink.as_ref()
955    }
956
957    #[allow(clippy::result_large_err)]
958    pub(crate) fn send(
959        &self,
960        event: NodeEndpointEvent,
961    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
962        if event.messages.is_empty() {
963            return Ok(());
964        }
965
966        self.send_event(event, true)
967    }
968
969    #[allow(clippy::result_large_err)]
970    fn send_event(
971        &self,
972        event: NodeEndpointEvent,
973        split_on_pressure: bool,
974    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
975        let count = event.message_count();
976        let Some(previous) =
977            try_reserve_endpoint_event_messages(&self.queued_messages, self.message_cap, count)
978        else {
979            if split_on_pressure && count > 1 {
980                return self.split_and_send_event(event);
981            }
982            crate::perf_profile::record_event_count(
983                crate::perf_profile::Event::EndpointEventBulkDropped,
984                count as u64,
985            );
986            return Ok(());
987        };
988
989        let queued = previous.saturating_add(count);
990        match self.tx.try_send(event) {
991            Ok(()) => {
992                self.note_send_success(previous, queued);
993                Ok(())
994            }
995            Err(tokio::sync::mpsc::error::TrySendError::Full(_event)) => {
996                self.note_send_rejected(count);
997                crate::perf_profile::record_event_count(
998                    crate::perf_profile::Event::EndpointEventBulkDropped,
999                    count as u64,
1000                );
1001                Ok(())
1002            }
1003            Err(tokio::sync::mpsc::error::TrySendError::Closed(event)) => {
1004                self.note_send_rejected(count);
1005                Err(tokio::sync::mpsc::error::SendError(event))
1006            }
1007        }
1008    }
1009
1010    #[allow(clippy::result_large_err)]
1011    fn split_and_send_event(
1012        &self,
1013        event: NodeEndpointEvent,
1014    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
1015        let mut messages = event.messages;
1016        let queued_at = event.queued_at;
1017        if messages.len() <= 1 {
1018            return self.send_event(
1019                NodeEndpointEvent {
1020                    messages,
1021                    queued_at,
1022                },
1023                false,
1024            );
1025        }
1026
1027        let right = messages.split_off(messages.len() / 2);
1028        if !messages.is_empty() {
1029            self.send_event(
1030                NodeEndpointEvent {
1031                    messages,
1032                    queued_at,
1033                },
1034                true,
1035            )?;
1036        }
1037        if !right.is_empty() {
1038            self.send_event(
1039                NodeEndpointEvent {
1040                    messages: right,
1041                    queued_at,
1042                },
1043                true,
1044            )?;
1045        }
1046        Ok(())
1047    }
1048
1049    fn note_send_success(&self, previous: usize, queued: usize) {
1050        if previous < ENDPOINT_EVENT_BACKLOG_HIGH_WATER
1051            && queued >= ENDPOINT_EVENT_BACKLOG_HIGH_WATER
1052        {
1053            crate::perf_profile::record_event(crate::perf_profile::Event::EndpointEventBacklogHigh);
1054        }
1055        self.ready.notify();
1056    }
1057
1058    fn note_send_rejected(&self, count: usize) {
1059        release_endpoint_event_messages(&self.queued_messages, count);
1060        self.ready.notify();
1061    }
1062
1063    #[cfg(test)]
1064    pub(crate) fn queued_messages(&self) -> usize {
1065        self.queued_messages.load(Relaxed)
1066    }
1067}
1068
1069impl Drop for EndpointEventSender {
1070    fn drop(&mut self) {
1071        self.ready.notify();
1072    }
1073}
1074
1075impl Drop for EndpointEventReceiver {
1076    fn drop(&mut self) {
1077        self.queued_messages.store(0, Relaxed);
1078        self.ready.notify();
1079    }
1080}
1081
1082impl EndpointEventRuntime {
1083    pub(in crate::node) fn attach(&mut self, sender: EndpointEventSender) {
1084        self.sender = Some(sender);
1085    }
1086
1087    pub(in crate::node) fn is_attached(&self) -> bool {
1088        self.sender.is_some()
1089    }
1090
1091    pub(in crate::node) fn sender(&self) -> Option<EndpointEventSender> {
1092        self.sender.clone()
1093    }
1094
1095    #[allow(clippy::result_large_err)]
1096    pub(in crate::node) fn deliver_endpoint_data_batch(
1097        &mut self,
1098        messages: Vec<EndpointDataDelivery>,
1099    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
1100        if messages.is_empty() {
1101            return Ok(());
1102        }
1103
1104        let Some(sender) = &self.sender else {
1105            return Ok(());
1106        };
1107        let _t_deliver =
1108            crate::perf_profile::Timer::start(crate::perf_profile::Stage::EndpointDeliver);
1109        sender.send(NodeEndpointEvent {
1110            messages,
1111            queued_at: crate::perf_profile::stamp(),
1112        })
1113    }
1114}
1115
1116impl EndpointEventReceiver {
1117    pub(crate) async fn recv(&mut self) -> Option<NodeEndpointEvent> {
1118        let event = self.rx.recv().await?;
1119        self.note_observed(&event);
1120        Some(event)
1121    }
1122
1123    pub(crate) fn blocking_recv(&mut self) -> Option<NodeEndpointEvent> {
1124        let mut observed = self.ready.snapshot();
1125        loop {
1126            match self.try_recv() {
1127                Ok(event) => return Some(event),
1128                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => return None,
1129                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
1130                    self.ready.wait_for_change(&mut observed);
1131                }
1132            }
1133        }
1134    }
1135
1136    pub(crate) fn try_recv(
1137        &mut self,
1138    ) -> Result<NodeEndpointEvent, tokio::sync::mpsc::error::TryRecvError> {
1139        match self.rx.try_recv() {
1140            Ok(event) => {
1141                self.note_observed(&event);
1142                Ok(event)
1143            }
1144            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
1145                if self.closed {
1146                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
1147                } else {
1148                    Err(tokio::sync::mpsc::error::TryRecvError::Empty)
1149                }
1150            }
1151            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
1152                self.closed = true;
1153                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
1154            }
1155        }
1156    }
1157
1158    pub(crate) fn release_messages(&self, count: usize) {
1159        release_endpoint_event_messages(&self.queued_messages, count);
1160    }
1161
1162    fn note_observed(&self, event: &NodeEndpointEvent) {
1163        event.record_dequeue_wait();
1164    }
1165}
1166
1167pub(in crate::node) fn release_endpoint_event_messages(counter: &AtomicUsize, count: usize) {
1168    if count == 0 {
1169        return;
1170    }
1171
1172    let previous = counter.fetch_sub(count, Relaxed);
1173    debug_assert!(
1174        previous >= count,
1175        "endpoint event queued message accounting underflow"
1176    );
1177}
1178
1179/// Reports what changed in response to `UpdatePeers`.
1180#[derive(Debug, Clone, Default, PartialEq, Eq)]
1181pub(crate) struct UpdatePeersOutcome {
1182    pub(crate) added: usize,
1183    pub(crate) removed: usize,
1184    pub(crate) updated: usize,
1185    pub(crate) unchanged: usize,
1186}
1187
1188/// Authenticated endpoint data emitted by the session receive path.
1189///
1190/// Keeping source identity and payload together makes the delivery-side
1191/// ownership boundary explicit for the current rx loop and for a future
1192/// peer/session runtime that can move endpoint-data delivery off the bounce path.
1193#[derive(Debug, Clone)]
1194pub(crate) struct EndpointDataDelivery {
1195    pub(crate) source_peer: PeerIdentity,
1196    pub(crate) payload: PacketBuffer,
1197    pub(crate) enqueued_at_ms: u64,
1198}
1199
1200impl EndpointDataDelivery {
1201    pub(crate) fn new(source_peer: PeerIdentity, payload: impl Into<PacketBuffer>) -> Self {
1202        Self {
1203            source_peer,
1204            payload: payload.into(),
1205            enqueued_at_ms: crate::time::now_ms(),
1206        }
1207    }
1208}
1209
1210/// Endpoint data events emitted by the node session receive path.
1211#[derive(Debug)]
1212pub(crate) struct NodeEndpointEvent {
1213    pub(crate) messages: Vec<EndpointDataDelivery>,
1214    pub(crate) queued_at: Option<crate::perf_profile::TraceStamp>,
1215}
1216
1217impl NodeEndpointEvent {
1218    pub(in crate::node) fn message_count(&self) -> usize {
1219        self.messages.len()
1220    }
1221
1222    fn queued_at(&self) -> Option<crate::perf_profile::TraceStamp> {
1223        self.queued_at
1224    }
1225
1226    fn record_dequeue_wait(&self) {
1227        let queued_at = self.queued_at();
1228        if queued_at.is_none() {
1229            return;
1230        }
1231        crate::perf_profile::record_since_count(
1232            crate::perf_profile::Stage::EndpointEventWait,
1233            queued_at,
1234            self.message_count() as u64,
1235        );
1236    }
1237}
1238
1239/// Authenticated peer state exposed to embedded endpoint callers.
1240#[derive(Debug, Clone, PartialEq, Eq)]
1241pub(crate) struct NodeEndpointPeer {
1242    pub(crate) npub: String,
1243    pub(crate) node_addr: NodeAddr,
1244    pub(crate) connected: bool,
1245    pub(crate) transport_addr: Option<String>,
1246    pub(crate) transport_type: Option<String>,
1247    pub(crate) link_id: u64,
1248    pub(crate) srtt_ms: Option<u64>,
1249    pub(crate) srtt_age_ms: Option<u64>,
1250    pub(crate) packets_sent: u64,
1251    pub(crate) packets_recv: u64,
1252    pub(crate) bytes_sent: u64,
1253    pub(crate) bytes_recv: u64,
1254    pub(crate) rekey_in_progress: bool,
1255    pub(crate) rekey_draining: bool,
1256    pub(crate) current_k_bit: Option<bool>,
1257    pub(crate) last_outbound_route: Option<String>,
1258    pub(crate) direct_probe_pending: bool,
1259    pub(crate) direct_probe_after_ms: Option<u64>,
1260    pub(crate) direct_probe_retry_count: u32,
1261    pub(crate) direct_probe_auto_reconnect: bool,
1262    pub(crate) direct_probe_expires_at_ms: Option<u64>,
1263    pub(crate) nostr_traversal_consecutive_failures: u32,
1264    pub(crate) nostr_traversal_in_cooldown: bool,
1265    pub(crate) nostr_traversal_cooldown_until_ms: Option<u64>,
1266    pub(crate) nostr_traversal_last_observed_skew_ms: Option<i64>,
1267}
1268
1269/// Live Nostr relay state exposed to embedded endpoint callers.
1270#[derive(Debug, Clone, PartialEq, Eq)]
1271pub(crate) struct NodeEndpointRelayStatus {
1272    pub(crate) url: String,
1273    pub(crate) status: String,
1274}
fips_core/node/endpoint_event.rs

fips_core/node/
endpoint_event.rs
