fips_core/node/
endpoint_event.rs

1use super::*;
2use crate::transport::PacketBuffer;
3#[cfg(unix)]
4use crossbeam_channel::{
5    Receiver as CrossbeamReceiver, Sender as CrossbeamSender, TryRecvError, TrySendError, bounded,
6};
7
8/// App-owned packet channels for embedding FIPS without a system TUN.
9#[derive(Debug)]
10pub struct ExternalPacketIo {
11    /// Send outbound IPv6 packets into the node.
12    pub outbound_tx: crate::upper::tun::TunOutboundTx,
13    /// Receive inbound IPv6 packets delivered by FIPS sessions.
14    pub inbound_rx: tokio::sync::mpsc::Receiver<NodeDeliveredPacket>,
15}
16
17/// App-owned endpoint data channels for embedding FIPS without a daemon.
18#[derive(Debug)]
19pub(crate) struct EndpointDataIo {
20    /// Send latency-sensitive endpoint data and management commands into the
21    /// node RX loop ahead of queued bulk endpoint data.
22    pub(crate) priority_command_tx: tokio::sync::mpsc::Sender<NodeEndpointCommand>,
23    /// Send endpoint data commands into the node RX loop.
24    ///
25    /// Bounded with a generous default so normal sender bursts do not
26    /// stall on semaphore acquisition. macOS pacing happens at the UDP
27    /// egress thread where the real Wi-Fi/interface bottleneck is visible;
28    /// constraining this app queue instead caused the inner TCP flow to
29    /// collapse under iperf. `FIPS_ENDPOINT_DATA_QUEUE_CAP` overrides the
30    /// default for benches.
31    pub(crate) command_tx: tokio::sync::mpsc::Sender<NodeEndpointCommand>,
32    /// Receive endpoint data delivered by FIPS sessions.
33    ///
34    /// Priority endpoint events use an unbounded lane so small control-shaped
35    /// packets keep a wait-free push from the rx loop. Bulk endpoint messages
36    /// are bounded by the endpoint-data capacity; oversized batches split at
37    /// the message-credit boundary before any remaining tail drops visibly via
38    /// `endpoint_event_bulk_dropped`. Backpressure is still visible through
39    /// `endpoint_event_wait` latency and `endpoint_event_backlog_high` when the
40    /// consumer falls materially behind.
41    pub(crate) event_rx: EndpointEventReceiver,
42    /// Clone of the event_tx exposed for in-process loopback (e.g.
43    /// `FipsEndpoint::send` to self_npub). Lets the endpoint inject an
44    /// event into the same queue without going through the encrypt /
45    /// decrypt path, while keeping every consumer reading from a single
46    /// channel.
47    pub(crate) event_tx: EndpointEventSender,
48    /// Shared endpoint-side bulk send runtime.
49    ///
50    /// The node publishes short-lived, invalidatable leases after it proves an
51    /// established UDP/worker route is usable. Endpoint bulk batches may use
52    /// those leases to prepare worker jobs without waiting for the rx-loop
53    /// command mailbox; priority/control packets keep using the command lane.
54    #[cfg(unix)]
55    pub(crate) bulk_send_runtime: EndpointBulkSendRuntime,
56}
57
58/// Shared endpoint-side bulk send lease store plus feedback lane.
59#[cfg(unix)]
60#[derive(Clone)]
61pub(crate) struct EndpointBulkSendRuntime {
62    leases: Arc<std::sync::RwLock<std::collections::HashMap<NodeAddr, EndpointBulkSendLease>>>,
63    feedback_tx: tokio::sync::mpsc::Sender<EndpointBulkSendFeedback>,
64    committed_dispatch: EndpointCommittedBulkDispatch,
65    generation: Arc<std::sync::atomic::AtomicU64>,
66}
67
68#[cfg(unix)]
69impl std::fmt::Debug for EndpointBulkSendRuntime {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        f.debug_struct("EndpointBulkSendRuntime")
72            .field("leases", &self.lease_count())
73            .finish_non_exhaustive()
74    }
75}
76
77#[cfg(unix)]
78#[derive(Clone)]
79pub(crate) struct EndpointBulkSendLease {
80    pub(in crate::node) source_addr: NodeAddr,
81    pub(in crate::node) dest_addr: NodeAddr,
82    pub(in crate::node) next_hop_addr: NodeAddr,
83    pub(in crate::node) path_mtu: u16,
84    pub(in crate::node) default_ttl: u8,
85    pub(in crate::node) scheduling_weight: u8,
86    pub(in crate::node) direct_path_blocks_direct_payload: bool,
87    pub(in crate::node) fsp: EndpointBulkSendFspLease,
88    pub(in crate::node) fmp: EndpointBulkSendFmpLease,
89    pub(in crate::node) send_target: crate::node::encrypt_worker::SelectedSendTarget,
90    pub(in crate::node) workers: crate::node::encrypt_worker::EncryptWorkerPool,
91    expires_at: crate::time::Instant,
92    generation: u64,
93}
94
95#[cfg(unix)]
96#[derive(Clone)]
97pub(crate) struct EndpointBulkSendFspLease {
98    pub(in crate::node) cipher: ring::aead::LessSafeKey,
99    pub(in crate::node) counter_authority: crate::noise::SendCounterAuthority,
100    pub(in crate::node) session_start_ms: u64,
101    pub(in crate::node) current_k_bit: bool,
102    pub(in crate::node) spin_bit: bool,
103}
104
105#[cfg(unix)]
106#[derive(Clone)]
107pub(crate) struct EndpointBulkSendFmpLease {
108    pub(in crate::node) cipher: ring::aead::LessSafeKey,
109    pub(in crate::node) counter_authority: crate::noise::SendCounterAuthority,
110    pub(in crate::node) their_index: crate::utils::index::SessionIndex,
111    pub(in crate::node) session_start: std::time::Instant,
112    pub(in crate::node) base_flags: u8,
113}
114
115#[cfg_attr(not(unix), allow(dead_code))]
116pub(crate) struct EndpointBulkSendFeedback {
117    pub(in crate::node) records: Vec<EndpointBulkSendFeedbackRecord>,
118}
119
120#[cfg_attr(not(unix), allow(dead_code))]
121#[derive(Clone, Copy)]
122pub(in crate::node) enum EndpointBulkSendSessionBookkeeping {
123    Fsp {
124        path_mtu: u16,
125        bookkeeping: FspSendBookkeepingInput,
126    },
127}
128
129#[cfg_attr(not(unix), allow(dead_code))]
130#[derive(Clone, Copy)]
131pub(crate) struct EndpointBulkSendFeedbackRecord {
132    pub(in crate::node) dest_addr: NodeAddr,
133    pub(in crate::node) next_hop_addr: NodeAddr,
134    pub(in crate::node) fmp_counter: u64,
135    pub(in crate::node) fmp_timestamp_ms: u32,
136    pub(in crate::node) fmp_wire_capacity: usize,
137    pub(in crate::node) originated_bytes: usize,
138    pub(in crate::node) session_bookkeeping: EndpointBulkSendSessionBookkeeping,
139}
140
141#[cfg(unix)]
142#[derive(Clone)]
143struct EndpointCommittedBulkDispatch {
144    tx: CrossbeamSender<EndpointCommittedBulkBatch>,
145}
146
147#[cfg(unix)]
148struct EndpointCommittedBulkBatch {
149    workers: crate::node::encrypt_worker::EncryptWorkerPool,
150    jobs: Vec<crate::node::encrypt_worker::FmpSendJob>,
151    ready: Arc<EndpointCommittedBulkReady>,
152}
153
154#[cfg(unix)]
155pub(in crate::node) struct EndpointCommittedBulkHandle {
156    ready: Option<Arc<EndpointCommittedBulkReady>>,
157}
158
159#[cfg(unix)]
160#[derive(Clone, Copy, Debug, Eq, PartialEq)]
161enum EndpointCommittedBulkState {
162    Pending,
163    Committed,
164    Canceled,
165}
166
167#[cfg(unix)]
168struct EndpointCommittedBulkReady {
169    state: std::sync::Mutex<EndpointCommittedBulkState>,
170    changed: Condvar,
171}
172
173#[cfg(unix)]
174impl EndpointCommittedBulkDispatch {
175    fn channel(capacity: usize) -> Self {
176        let (tx, rx) = bounded(capacity.max(1));
177        std::thread::Builder::new()
178            .name("fips-endpoint-bulk-commit".to_string())
179            .spawn(move || run_endpoint_committed_bulk_dispatch(rx))
180            .expect("failed to spawn FIPS endpoint committed bulk dispatcher");
181        Self { tx }
182    }
183
184    fn try_stage(
185        &self,
186        workers: crate::node::encrypt_worker::EncryptWorkerPool,
187        jobs: Vec<crate::node::encrypt_worker::FmpSendJob>,
188    ) -> Option<EndpointCommittedBulkHandle> {
189        if jobs.is_empty() {
190            return Some(EndpointCommittedBulkHandle { ready: None });
191        }
192
193        let ready = Arc::new(EndpointCommittedBulkReady::new());
194        let batch = EndpointCommittedBulkBatch {
195            workers,
196            jobs,
197            ready: Arc::clone(&ready),
198        };
199        match self.tx.try_send(batch) {
200            Ok(()) => Some(EndpointCommittedBulkHandle { ready: Some(ready) }),
201            Err(TrySendError::Full(batch)) | Err(TrySendError::Disconnected(batch)) => {
202                batch.ready.cancel();
203                None
204            }
205        }
206    }
207}
208
209#[cfg(unix)]
210impl EndpointCommittedBulkBatch {
211    fn packet_count(&self) -> usize {
212        self.jobs.len()
213    }
214
215    fn can_merge_after(&self, other: &Self) -> bool {
216        crate::node::encrypt_worker::fmp_send_job_batches_share_bulk_target(&self.jobs, &other.jobs)
217    }
218
219    fn append_committed(&mut self, mut other: Self) {
220        self.jobs.append(&mut other.jobs);
221    }
222}
223
224#[cfg(unix)]
225impl EndpointCommittedBulkHandle {
226    pub(in crate::node) fn commit(mut self) {
227        if let Some(ready) = self.ready.take() {
228            ready.commit();
229        }
230    }
231
232    pub(in crate::node) fn cancel(mut self) {
233        if let Some(ready) = self.ready.take() {
234            ready.cancel();
235        }
236    }
237}
238
239#[cfg(unix)]
240impl Drop for EndpointCommittedBulkHandle {
241    fn drop(&mut self) {
242        if let Some(ready) = self.ready.take() {
243            ready.cancel();
244        }
245    }
246}
247
248#[cfg(unix)]
249impl EndpointCommittedBulkReady {
250    fn new() -> Self {
251        Self {
252            state: std::sync::Mutex::new(EndpointCommittedBulkState::Pending),
253            changed: Condvar::new(),
254        }
255    }
256
257    fn commit(&self) {
258        self.complete(EndpointCommittedBulkState::Committed);
259    }
260
261    fn cancel(&self) {
262        self.complete(EndpointCommittedBulkState::Canceled);
263    }
264
265    fn complete(&self, next: EndpointCommittedBulkState) {
266        let Ok(mut state) = self.state.lock() else {
267            return;
268        };
269        if *state == EndpointCommittedBulkState::Pending {
270            *state = next;
271            self.changed.notify_one();
272        }
273    }
274
275    fn wait(&self) -> EndpointCommittedBulkState {
276        let Ok(mut state) = self.state.lock() else {
277            return EndpointCommittedBulkState::Canceled;
278        };
279        while *state == EndpointCommittedBulkState::Pending {
280            match self.changed.wait(state) {
281                Ok(next) => state = next,
282                Err(_) => return EndpointCommittedBulkState::Canceled,
283            }
284        }
285        *state
286    }
287
288    fn try_state(&self) -> EndpointCommittedBulkState {
289        self.state
290            .lock()
291            .map(|state| *state)
292            .unwrap_or(EndpointCommittedBulkState::Canceled)
293    }
294}
295
296#[cfg(unix)]
297fn run_endpoint_committed_bulk_dispatch(rx: CrossbeamReceiver<EndpointCommittedBulkBatch>) {
298    let mut pending: Option<EndpointCommittedBulkBatch> = None;
299    loop {
300        let Some(mut batch) = pending.take().or_else(|| rx.recv().ok()) else {
301            break;
302        };
303        if batch.ready.wait() != EndpointCommittedBulkState::Committed {
304            continue;
305        }
306
307        let mut merged_batches = 0usize;
308        let mut merged_packets = 0usize;
309        let max_batches = endpoint_committed_bulk_coalesce_batches();
310        let max_packets = endpoint_committed_bulk_coalesce_packets();
311        // Merge only already-committed adjacent bulk batches. Pending or
312        // different-target work is held for the next loop, preserving FIFO
313        // without widening UDP_GSO bursts or waiting on future commits.
314        while merged_batches + 1 < max_batches && batch.packet_count() < max_packets {
315            match rx.try_recv() {
316                Ok(next) => match next.ready.try_state() {
317                    EndpointCommittedBulkState::Committed => {
318                        if !batch.can_merge_after(&next) {
319                            pending = Some(next);
320                            break;
321                        }
322                        if batch.packet_count().saturating_add(next.packet_count()) > max_packets {
323                            pending = Some(next);
324                            break;
325                        }
326                        merged_batches = merged_batches.saturating_add(1);
327                        merged_packets = merged_packets.saturating_add(next.packet_count());
328                        batch.append_committed(next);
329                    }
330                    EndpointCommittedBulkState::Canceled => {}
331                    EndpointCommittedBulkState::Pending => {
332                        pending = Some(next);
333                        break;
334                    }
335                },
336                Err(TryRecvError::Empty) => break,
337                Err(TryRecvError::Disconnected) => break,
338            }
339        }
340
341        crate::perf_profile::record_endpoint_committed_bulk_dispatch(
342            batch.packet_count(),
343            merged_batches,
344            merged_packets,
345        );
346        let EndpointCommittedBulkBatch { workers, jobs, .. } = batch;
347        let _all_enqueued = workers.dispatch_bulk_batch_blocking(jobs);
348    }
349}
350
351#[cfg(unix)]
352fn endpoint_committed_bulk_coalesce_batches() -> usize {
353    static VALUE: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
354    *VALUE.get_or_init(|| {
355        std::env::var("FIPS_ENDPOINT_COMMITTED_BULK_COALESCE_BATCHES")
356            .ok()
357            .and_then(|raw| raw.trim().parse::<usize>().ok())
358            // Default off: the committed batches are usually large already;
359            // adjacent coalescing is kept as an explicit benchmark knob.
360            .unwrap_or(1)
361            .clamp(1, 16)
362    })
363}
364
365#[cfg(unix)]
366fn endpoint_committed_bulk_coalesce_packets() -> usize {
367    static VALUE: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
368    *VALUE.get_or_init(|| {
369        std::env::var("FIPS_ENDPOINT_COMMITTED_BULK_COALESCE_PACKETS")
370            .ok()
371            .and_then(|raw| raw.trim().parse::<usize>().ok())
372            .unwrap_or(128)
373            .clamp(1, 1024)
374    })
375}
376
377#[cfg(unix)]
378impl EndpointBulkSendRuntime {
379    pub(in crate::node) fn channel(
380        capacity: usize,
381    ) -> (Self, tokio::sync::mpsc::Receiver<EndpointBulkSendFeedback>) {
382        let feedback_capacity = endpoint_data_command_capacity(capacity).max(1);
383        let (feedback_tx, feedback_rx) = tokio::sync::mpsc::channel(feedback_capacity);
384        let committed_dispatch = EndpointCommittedBulkDispatch::channel(capacity);
385        (
386            Self {
387                leases: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())),
388                feedback_tx,
389                committed_dispatch,
390                generation: Arc::new(std::sync::atomic::AtomicU64::new(1)),
391            },
392            feedback_rx,
393        )
394    }
395
396    pub(in crate::node) fn publish(&self, mut lease: EndpointBulkSendLease) {
397        lease.generation = self.generation.load(Relaxed);
398        if let Ok(mut leases) = self.leases.write() {
399            leases.insert(lease.dest_addr, lease);
400        }
401    }
402
403    pub(in crate::node) fn invalidate(&self, dest_addr: &NodeAddr) {
404        self.generation
405            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
406        if let Ok(mut leases) = self.leases.write() {
407            leases.remove(dest_addr);
408        }
409    }
410
411    pub(in crate::node) fn lease(&self, dest_addr: &NodeAddr) -> Option<EndpointBulkSendLease> {
412        let now = crate::time::instant_now();
413        let generation = self.generation.load(Relaxed);
414        let mut expired = false;
415        let lease = self.leases.read().ok().and_then(|leases| {
416            leases.get(dest_addr).and_then(|lease| {
417                if lease.generation != generation || lease.expires_at <= now {
418                    expired = true;
419                    None
420                } else {
421                    Some(lease.clone())
422                }
423            })
424        });
425        if expired && let Ok(mut leases) = self.leases.write() {
426            leases.remove(dest_addr);
427        }
428        lease
429    }
430
431    pub(in crate::node) fn try_feedback(
432        &self,
433        records: Vec<EndpointBulkSendFeedbackRecord>,
434    ) -> bool {
435        if records.is_empty() {
436            return true;
437        }
438        self.feedback_tx
439            .try_send(EndpointBulkSendFeedback { records })
440            .is_ok()
441    }
442
443    pub(in crate::node) fn try_stage_committed_bulk_dispatch(
444        &self,
445        workers: crate::node::encrypt_worker::EncryptWorkerPool,
446        jobs: Vec<crate::node::encrypt_worker::FmpSendJob>,
447    ) -> Option<EndpointCommittedBulkHandle> {
448        self.committed_dispatch.try_stage(workers, jobs)
449    }
450
451    fn lease_count(&self) -> usize {
452        self.leases.read().map(|leases| leases.len()).unwrap_or(0)
453    }
454}
455
456#[cfg(unix)]
457impl EndpointBulkSendLease {
458    pub(in crate::node) fn new(
459        source_addr: NodeAddr,
460        dest_addr: NodeAddr,
461        next_hop_addr: NodeAddr,
462        path_mtu: u16,
463        default_ttl: u8,
464        scheduling_weight: u8,
465        direct_path_blocks_direct_payload: bool,
466        fsp: EndpointBulkSendFspLease,
467        fmp: EndpointBulkSendFmpLease,
468        send_target: crate::node::encrypt_worker::SelectedSendTarget,
469        workers: crate::node::encrypt_worker::EncryptWorkerPool,
470        ttl: std::time::Duration,
471    ) -> Self {
472        Self {
473            source_addr,
474            dest_addr,
475            next_hop_addr,
476            path_mtu,
477            default_ttl,
478            scheduling_weight,
479            direct_path_blocks_direct_payload,
480            fsp,
481            fmp,
482            send_target,
483            workers,
484            expires_at: crate::time::instant_now() + ttl,
485            generation: 0,
486        }
487    }
488}
489
490/// Observable owner for endpoint events delivered to embedded applications.
491#[derive(Debug, Clone)]
492pub(crate) struct EndpointEventSender {
493    priority: tokio::sync::mpsc::UnboundedSender<NodeEndpointEvent>,
494    bulk: tokio::sync::mpsc::Sender<NodeEndpointEvent>,
495    queued_messages: Arc<AtomicUsize>,
496    bulk_queued_messages: Arc<AtomicUsize>,
497    ready: Arc<EndpointEventReady>,
498    bulk_message_cap: usize,
499}
500
501#[derive(Debug)]
502pub(crate) struct EndpointEventReceiver {
503    priority: tokio::sync::mpsc::UnboundedReceiver<NodeEndpointEvent>,
504    bulk: tokio::sync::mpsc::Receiver<NodeEndpointEvent>,
505    queued_messages: Arc<AtomicUsize>,
506    bulk_queued_messages: Arc<AtomicUsize>,
507    ready: Arc<EndpointEventReady>,
508    priority_closed: bool,
509    bulk_closed: bool,
510}
511
512#[derive(Debug, Default)]
513struct EndpointEventReady {
514    sequence: StdMutex<u64>,
515    changed: Condvar,
516}
517
518impl EndpointEventReady {
519    fn notify(&self) {
520        if let Ok(mut sequence) = self.sequence.lock() {
521            *sequence = sequence.wrapping_add(1);
522            self.changed.notify_one();
523        }
524    }
525
526    fn snapshot(&self) -> u64 {
527        self.sequence.lock().map(|sequence| *sequence).unwrap_or(0)
528    }
529
530    fn wait_for_change(&self, observed: &mut u64) {
531        let Ok(mut sequence) = self.sequence.lock() else {
532            return;
533        };
534        while *sequence == *observed {
535            match self.changed.wait(sequence) {
536                Ok(next) => sequence = next,
537                Err(_) => return,
538            }
539        }
540        *observed = *sequence;
541    }
542}
543
544#[derive(Clone, Copy)]
545enum EndpointEventLane {
546    Priority,
547    Bulk,
548}
549
550fn endpoint_event_lane_for_len(len: usize) -> EndpointEventLane {
551    if len <= ENDPOINT_EVENT_PRIORITY_MAX_LEN {
552        EndpointEventLane::Priority
553    } else {
554        EndpointEventLane::Bulk
555    }
556}
557
558fn endpoint_event_bulk_capacity(requested: usize) -> usize {
559    requested.max(1)
560}
561
562fn try_reserve_endpoint_event_bulk_messages(
563    counter: &AtomicUsize,
564    capacity: usize,
565    count: usize,
566) -> Option<usize> {
567    if count == 0 {
568        return Some(counter.load(Relaxed));
569    }
570
571    counter
572        .fetch_update(Relaxed, Relaxed, |current| {
573            current.checked_add(count).filter(|next| *next <= capacity)
574        })
575        .ok()
576}
577
578/// Delivery-side owner for endpoint data emitted by session receive handling.
579///
580/// The rx loop currently owns this runtime, but keeping sender, batching, and
581/// backlog accounting behind one value makes the future peer/shard receive
582/// runtime move explicit instead of threading endpoint-event fields through
583/// `Node` packet handlers.
584#[derive(Debug, Default)]
585pub(in crate::node) struct EndpointEventRuntime {
586    sender: Option<EndpointEventSender>,
587    batch_depth: usize,
588    batch: Vec<EndpointDataDelivery>,
589}
590
591impl EndpointEventSender {
592    pub(in crate::node) fn channel(capacity: usize) -> (Self, EndpointEventReceiver) {
593        let (priority_tx, priority_rx) = tokio::sync::mpsc::unbounded_channel();
594        let bulk_message_cap = endpoint_event_bulk_capacity(capacity);
595        let (bulk_tx, bulk_rx) = tokio::sync::mpsc::channel(bulk_message_cap);
596        let queued_messages = Arc::new(AtomicUsize::new(0));
597        let bulk_queued_messages = Arc::new(AtomicUsize::new(0));
598        let ready = Arc::new(EndpointEventReady::default());
599        (
600            Self {
601                priority: priority_tx,
602                bulk: bulk_tx,
603                queued_messages: Arc::clone(&queued_messages),
604                bulk_queued_messages: Arc::clone(&bulk_queued_messages),
605                ready: Arc::clone(&ready),
606                bulk_message_cap,
607            },
608            EndpointEventReceiver {
609                priority: priority_rx,
610                bulk: bulk_rx,
611                queued_messages,
612                bulk_queued_messages,
613                ready,
614                priority_closed: false,
615                bulk_closed: false,
616            },
617        )
618    }
619
620    pub(in crate::node) fn same_channels(&self, other: &Self) -> bool {
621        self.priority.same_channel(&other.priority)
622            && self.bulk.same_channel(&other.bulk)
623            && Arc::ptr_eq(&self.queued_messages, &other.queued_messages)
624            && Arc::ptr_eq(&self.bulk_queued_messages, &other.bulk_queued_messages)
625            && Arc::ptr_eq(&self.ready, &other.ready)
626            && self.bulk_message_cap == other.bulk_message_cap
627    }
628
629    #[allow(clippy::result_large_err)]
630    pub(crate) fn send(
631        &self,
632        event: NodeEndpointEvent,
633    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
634        match event {
635            NodeEndpointEvent::Data {
636                source_peer,
637                payload,
638                queued_at,
639            } => {
640                let lane = endpoint_event_lane_for_len(payload.len());
641                self.send_to_lane(
642                    NodeEndpointEvent::Data {
643                        source_peer,
644                        payload,
645                        queued_at,
646                    },
647                    lane,
648                )
649            }
650            NodeEndpointEvent::DataBatch {
651                messages,
652                queued_at,
653            } => self.send_data_batch(messages, queued_at),
654        }
655    }
656
657    #[allow(clippy::result_large_err)]
658    fn send_data_batch(
659        &self,
660        messages: Vec<EndpointDataDelivery>,
661        queued_at: Option<crate::perf_profile::TraceStamp>,
662    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
663        if messages.is_empty() {
664            return Ok(());
665        }
666
667        let message_count = messages.len();
668        let priority_count = messages
669            .iter()
670            .filter(|message| message.is_priority_sized())
671            .count();
672        if priority_count == 0 || priority_count == message_count {
673            let lane = if priority_count == 0 {
674                EndpointEventLane::Bulk
675            } else {
676                EndpointEventLane::Priority
677            };
678            let event = NodeEndpointEvent::from_delivery_messages(messages, queued_at)
679                .expect("non-empty endpoint event batch should produce event");
680            return self.send_to_lane(event, lane);
681        }
682
683        let mut priority_messages = Vec::with_capacity(priority_count);
684        let mut bulk_messages = Vec::with_capacity(message_count - priority_count);
685        for message in messages {
686            if message.is_priority_sized() {
687                priority_messages.push(message);
688            } else {
689                bulk_messages.push(message);
690            }
691        }
692
693        if let Some(event) = NodeEndpointEvent::from_delivery_messages(priority_messages, queued_at)
694        {
695            self.send_to_lane(event, EndpointEventLane::Priority)?;
696        }
697        if let Some(event) = NodeEndpointEvent::from_delivery_messages(bulk_messages, queued_at) {
698            self.send_to_lane(event, EndpointEventLane::Bulk)?;
699        }
700        Ok(())
701    }
702
703    #[allow(clippy::result_large_err)]
704    fn send_to_lane(
705        &self,
706        event: NodeEndpointEvent,
707        lane: EndpointEventLane,
708    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
709        if matches!(lane, EndpointEventLane::Bulk) {
710            return self.send_bulk_to_lane(event, true);
711        }
712
713        let count = event.message_count();
714        let previous = self.queued_messages.fetch_add(count, Relaxed);
715        let queued = previous.saturating_add(count);
716        match self.priority.send(event) {
717            Ok(()) => {
718                self.note_send_success(previous, queued);
719                Ok(())
720            }
721            Err(error) => {
722                self.note_send_rejected(count);
723                Err(error)
724            }
725        }
726    }
727
728    #[allow(clippy::result_large_err)]
729    fn send_bulk_to_lane(
730        &self,
731        event: NodeEndpointEvent,
732        split_on_pressure: bool,
733    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
734        let count = event.message_count();
735        let Some(previous_bulk) = try_reserve_endpoint_event_bulk_messages(
736            &self.bulk_queued_messages,
737            self.bulk_message_cap,
738            count,
739        ) else {
740            if split_on_pressure && count > 1 {
741                return self.split_and_send_bulk_event(event);
742            }
743            crate::perf_profile::record_event_count(
744                crate::perf_profile::Event::EndpointEventBulkDropped,
745                count as u64,
746            );
747            return Ok(());
748        };
749
750        let queued_bulk = previous_bulk.saturating_add(count);
751        if previous_bulk < ENDPOINT_EVENT_BACKLOG_HIGH_WATER
752            && queued_bulk >= ENDPOINT_EVENT_BACKLOG_HIGH_WATER
753        {
754            crate::perf_profile::record_event(
755                crate::perf_profile::Event::EndpointEventBulkBacklogHigh,
756            );
757        }
758
759        let previous = self.queued_messages.fetch_add(count, Relaxed);
760        let queued = previous.saturating_add(count);
761        match self.bulk.try_send(event) {
762            Ok(()) => {
763                self.note_send_success(previous, queued);
764                Ok(())
765            }
766            Err(tokio::sync::mpsc::error::TrySendError::Full(_event)) => {
767                self.note_bulk_send_rejected(count);
768                crate::perf_profile::record_event_count(
769                    crate::perf_profile::Event::EndpointEventBulkDropped,
770                    count as u64,
771                );
772                Ok(())
773            }
774            Err(tokio::sync::mpsc::error::TrySendError::Closed(event)) => {
775                self.note_bulk_send_rejected(count);
776                Err(tokio::sync::mpsc::error::SendError(event))
777            }
778        }
779    }
780
781    #[allow(clippy::result_large_err)]
782    fn split_and_send_bulk_event(
783        &self,
784        event: NodeEndpointEvent,
785    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
786        let (mut messages, queued_at) = match event {
787            NodeEndpointEvent::DataBatch {
788                messages,
789                queued_at,
790            } => (messages, queued_at),
791            event => {
792                let count = event.message_count();
793                crate::perf_profile::record_event_count(
794                    crate::perf_profile::Event::EndpointEventBulkDropped,
795                    count as u64,
796                );
797                return Ok(());
798            }
799        };
800        if messages.len() <= 1 {
801            let event = NodeEndpointEvent::from_delivery_messages(messages, queued_at)
802                .expect("non-empty split endpoint batch should produce an event");
803            return self.send_bulk_to_lane(event, false);
804        }
805
806        let right = messages.split_off(messages.len() / 2);
807        if let Some(left) = NodeEndpointEvent::from_delivery_messages(messages, queued_at) {
808            self.send_bulk_to_lane(left, true)?;
809        }
810        if let Some(right) = NodeEndpointEvent::from_delivery_messages(right, queued_at) {
811            self.send_bulk_to_lane(right, true)?;
812        }
813        Ok(())
814    }
815
816    fn note_send_success(&self, previous: usize, queued: usize) {
817        if previous < ENDPOINT_EVENT_BACKLOG_HIGH_WATER
818            && queued >= ENDPOINT_EVENT_BACKLOG_HIGH_WATER
819        {
820            crate::perf_profile::record_event(crate::perf_profile::Event::EndpointEventBacklogHigh);
821        }
822        self.ready.notify();
823    }
824
825    fn note_send_rejected(&self, count: usize) {
826        release_endpoint_event_messages(&self.queued_messages, count);
827        self.ready.notify();
828    }
829
830    fn note_bulk_send_rejected(&self, count: usize) {
831        release_endpoint_event_messages(&self.queued_messages, count);
832        release_endpoint_event_messages(&self.bulk_queued_messages, count);
833        self.ready.notify();
834    }
835
836    #[cfg(test)]
837    pub(crate) fn queued_messages(&self) -> usize {
838        self.queued_messages.load(Relaxed)
839    }
840
841    #[cfg(test)]
842    pub(crate) fn bulk_queued_messages(&self) -> usize {
843        self.bulk_queued_messages.load(Relaxed)
844    }
845}
846
847impl Drop for EndpointEventSender {
848    fn drop(&mut self) {
849        self.ready.notify();
850    }
851}
852
853impl EndpointEventRuntime {
854    pub(in crate::node) fn attach(&mut self, sender: EndpointEventSender) {
855        self.sender = Some(sender);
856        self.batch_depth = 0;
857        self.batch.clear();
858    }
859
860    pub(in crate::node) fn is_attached(&self) -> bool {
861        self.sender.is_some()
862    }
863
864    pub(in crate::node) fn sender(&self) -> Option<EndpointEventSender> {
865        self.sender.clone()
866    }
867
868    pub(in crate::node) fn begin_batch(&mut self) {
869        if self.is_attached() {
870            self.batch_depth = self.batch_depth.saturating_add(1);
871        }
872    }
873
874    pub(in crate::node) fn finish_batch(&mut self) {
875        if self.batch_depth == 0 {
876            return;
877        }
878        self.batch_depth -= 1;
879        if self.batch_depth == 0 {
880            self.flush_batch();
881        }
882    }
883
884    #[allow(clippy::result_large_err)]
885    pub(in crate::node) fn deliver_endpoint_data(
886        &mut self,
887        message: EndpointDataDelivery,
888    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
889        if self.batch_depth > 0 {
890            self.batch.push(message);
891            return Ok(());
892        }
893
894        self.send(NodeEndpointEvent::Data {
895            source_peer: message.source_peer,
896            payload: message.payload,
897            queued_at: crate::perf_profile::stamp(),
898        })
899    }
900
901    fn flush_batch(&mut self) {
902        let count = self.batch.len();
903        if count == 0 {
904            return;
905        }
906
907        let queued_at = crate::perf_profile::stamp();
908        let event = if count == 1 {
909            let message = self.batch.pop().expect("batch should contain message");
910            NodeEndpointEvent::Data {
911                source_peer: message.source_peer,
912                payload: message.payload,
913                queued_at,
914            }
915        } else {
916            NodeEndpointEvent::DataBatch {
917                messages: std::mem::take(&mut self.batch),
918                queued_at,
919            }
920        };
921
922        if let Err(error) = self.send(event) {
923            debug!(
924                error = %error,
925                messages = count,
926                "Failed to deliver endpoint data event batch"
927            );
928        }
929    }
930
931    #[allow(clippy::result_large_err)]
932    fn send(
933        &self,
934        event: NodeEndpointEvent,
935    ) -> Result<(), tokio::sync::mpsc::error::SendError<NodeEndpointEvent>> {
936        let Some(sender) = &self.sender else {
937            return Ok(());
938        };
939        let _t_deliver =
940            crate::perf_profile::Timer::start(crate::perf_profile::Stage::EndpointDeliver);
941        sender.send(event)
942    }
943}
944
945impl EndpointEventReceiver {
946    pub(crate) async fn recv(&mut self) -> Option<NodeEndpointEvent> {
947        loop {
948            match self.try_recv() {
949                Ok(event) => return Some(event),
950                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => return None,
951                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {}
952            }
953
954            tokio::select! {
955                biased;
956                event = self.priority.recv(), if !self.priority_closed => {
957                    match event {
958                        Some(event) => {
959                            self.note_dequeued(&event);
960                            return Some(event);
961                        }
962                        None => self.priority_closed = true,
963                    }
964                }
965                event = self.bulk.recv(), if !self.bulk_closed => {
966                    match event {
967                        Some(event) => {
968                            self.note_dequeued(&event);
969                            return Some(event);
970                        }
971                        None => self.bulk_closed = true,
972                    }
973                }
974            }
975        }
976    }
977
978    pub(crate) fn blocking_recv(&mut self) -> Option<NodeEndpointEvent> {
979        let mut observed = self.ready.snapshot();
980        loop {
981            match self.try_recv() {
982                Ok(event) => return Some(event),
983                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => return None,
984                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
985                    self.ready.wait_for_change(&mut observed);
986                }
987            }
988        }
989    }
990
991    pub(crate) fn try_recv(
992        &mut self,
993    ) -> Result<NodeEndpointEvent, tokio::sync::mpsc::error::TryRecvError> {
994        match self.try_recv_priority() {
995            Ok(event) => return Ok(event),
996            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {}
997            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {}
998        }
999
1000        match self.bulk.try_recv() {
1001            Ok(event) => {
1002                self.note_dequeued(&event);
1003                Ok(event)
1004            }
1005            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
1006                if self.priority_closed && self.bulk_closed {
1007                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
1008                } else {
1009                    Err(tokio::sync::mpsc::error::TryRecvError::Empty)
1010                }
1011            }
1012            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
1013                self.bulk_closed = true;
1014                if self.priority_closed {
1015                    Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
1016                } else {
1017                    Err(tokio::sync::mpsc::error::TryRecvError::Empty)
1018                }
1019            }
1020        }
1021    }
1022
1023    pub(crate) fn try_recv_priority(
1024        &mut self,
1025    ) -> Result<NodeEndpointEvent, tokio::sync::mpsc::error::TryRecvError> {
1026        let event = match self.priority.try_recv() {
1027            Ok(event) => event,
1028            Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
1029                return Err(tokio::sync::mpsc::error::TryRecvError::Empty);
1030            }
1031            Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
1032                self.priority_closed = true;
1033                return Err(tokio::sync::mpsc::error::TryRecvError::Disconnected);
1034            }
1035        };
1036        self.note_dequeued(&event);
1037        Ok(event)
1038    }
1039
1040    fn note_dequeued(&self, event: &NodeEndpointEvent) {
1041        event.record_dequeue_wait();
1042        let counts = event.dequeue_counts();
1043        release_endpoint_event_messages(&self.queued_messages, counts.total);
1044        release_endpoint_event_messages(&self.bulk_queued_messages, counts.bulk);
1045    }
1046}
1047
1048pub(in crate::node) fn release_endpoint_event_messages(counter: &AtomicUsize, count: usize) {
1049    if count == 0 {
1050        return;
1051    }
1052
1053    let previous = counter.fetch_sub(count, Relaxed);
1054    debug_assert!(
1055        previous >= count,
1056        "endpoint event queued message accounting underflow"
1057    );
1058}
1059
1060pub(crate) fn endpoint_data_command_capacity(requested: usize) -> usize {
1061    if let Ok(raw) = std::env::var("FIPS_ENDPOINT_DATA_QUEUE_CAP")
1062        && let Ok(value) = raw.trim().parse::<usize>()
1063        && value > 0
1064    {
1065        return value;
1066    }
1067
1068    requested.max(1).max(32_768)
1069}
1070
1071// Endpoint send batches have already paid the per-packet mpsc wakeup and peer
1072// identity costs at the embedded API boundary. Charge rx_loop drain budget in
1073// small packet groups so full batches keep moving without letting one hot
1074// endpoint queue monopolize the coordinator.
1075const ENDPOINT_SEND_BATCH_DRAIN_QUANTUM: usize = 8;
1076
1077fn endpoint_send_batch_drain_cost(packet_count: usize) -> usize {
1078    packet_count
1079        .max(1)
1080        .saturating_add(ENDPOINT_SEND_BATCH_DRAIN_QUANTUM - 1)
1081        / ENDPOINT_SEND_BATCH_DRAIN_QUANTUM
1082}
1083
1084/// Commands accepted by the node endpoint data service.
1085#[derive(Debug)]
1086pub(crate) enum NodeEndpointCommand {
1087    /// Send with an explicit response channel — used by callers that
1088    /// care whether the local-stack handoff succeeded (e.g.
1089    /// `blocking_send` waits for the runtime to accept the send).
1090    Send {
1091        command: EndpointSendCommand,
1092        response_tx: tokio::sync::oneshot::Sender<Result<(), NodeError>>,
1093    },
1094    /// **Fire-and-forget** variant of `Send` — no oneshot allocation,
1095    /// no per-packet result channel. Used by the data-plane fast path
1096    /// (`FipsEndpoint::send`) where the caller already discards the
1097    /// result. Saves one oneshot::channel() allocation per outbound
1098    /// packet on the application's send hot path.
1099    SendOneway { command: EndpointSendCommand },
1100    /// Fire-and-forget batch of endpoint payloads that already share the same
1101    /// peer and command lane. This keeps bursty embedded dataplanes from
1102    /// paying one mpsc send/wake per packet while preserving the priority/bulk
1103    /// split without repeating the resolved peer identity in every payload.
1104    SendBatchOneway {
1105        command: EndpointSendBatchCommand,
1106        lane: EndpointCommandLane,
1107    },
1108    PeerSnapshot {
1109        response_tx: tokio::sync::oneshot::Sender<Vec<NodeEndpointPeer>>,
1110    },
1111    RelaySnapshot {
1112        response_tx: tokio::sync::oneshot::Sender<Vec<NodeEndpointRelayStatus>>,
1113    },
1114    UpdateRelays {
1115        advert_relays: Vec<String>,
1116        dm_relays: Vec<String>,
1117        response_tx: tokio::sync::oneshot::Sender<Result<(), NodeError>>,
1118    },
1119    /// Replace the runtime peer list. Newly added auto-connect peers get
1120    /// `initiate_peer_connection` immediately; removed peers are dropped
1121    /// from the retry queue (the regular liveness timeout reaps any active
1122    /// session). Existing entries are kept and their `addresses` field is
1123    /// refreshed so the next retry sees the latest hints.
1124    UpdatePeers {
1125        peers: Vec<crate::config::PeerConfig>,
1126        response_tx: tokio::sync::oneshot::Sender<Result<UpdatePeersOutcome, NodeError>>,
1127    },
1128    /// Force immediate direct-path refresh attempts for configured peers.
1129    ///
1130    /// This is intentionally separate from `UpdatePeers`: callers may need to
1131    /// reprobe a stale active path even when the configured address set is
1132    /// unchanged, and `update_peers` avoids churning an active peer that is
1133    /// already on a known candidate.
1134    RefreshPeerPaths {
1135        npubs: Vec<String>,
1136        response_tx: tokio::sync::oneshot::Sender<Result<usize, NodeError>>,
1137    },
1138}
1139
1140/// Message payload for outbound endpoint data handed from an embedded
1141/// application into the node rx loop.
1142#[derive(Debug)]
1143pub(crate) struct EndpointSendCommand {
1144    send: EndpointDataSend,
1145    queued_at: Option<crate::perf_profile::TraceStamp>,
1146}
1147
1148impl EndpointSendCommand {
1149    pub(crate) fn new(
1150        remote: PeerIdentity,
1151        payload: Vec<u8>,
1152        queued_at: Option<crate::perf_profile::TraceStamp>,
1153    ) -> Self {
1154        Self {
1155            send: EndpointDataSend::new(remote, EndpointDataPayload::new(payload)),
1156            queued_at,
1157        }
1158    }
1159
1160    pub(crate) fn lane(&self) -> EndpointCommandLane {
1161        self.send.payload().lane()
1162    }
1163
1164    pub(crate) fn drop_on_backpressure(&self) -> bool {
1165        self.send.payload().drop_on_backpressure()
1166    }
1167
1168    pub(crate) fn into_parts(self) -> (EndpointDataSend, Option<crate::perf_profile::TraceStamp>) {
1169        (self.send, self.queued_at)
1170    }
1171}
1172
1173/// Batch of endpoint payloads to one resolved peer.
1174#[derive(Debug)]
1175pub(crate) struct EndpointSendBatchCommand {
1176    remote: PeerIdentity,
1177    payloads: Vec<EndpointDataPayload>,
1178    queued_at: Option<crate::perf_profile::TraceStamp>,
1179}
1180
1181impl EndpointSendBatchCommand {
1182    pub(crate) fn new(
1183        remote: PeerIdentity,
1184        payloads: Vec<EndpointDataPayload>,
1185        queued_at: Option<crate::perf_profile::TraceStamp>,
1186    ) -> Option<Self> {
1187        if payloads.is_empty() {
1188            return None;
1189        }
1190        Some(Self {
1191            remote,
1192            payloads,
1193            queued_at,
1194        })
1195    }
1196
1197    pub(crate) fn lane(&self) -> EndpointCommandLane {
1198        self.payloads[0].lane()
1199    }
1200
1201    pub(crate) fn len(&self) -> usize {
1202        self.payloads.len()
1203    }
1204
1205    pub(crate) fn can_coalesce_with(&self, other: &Self, max_payloads: usize) -> bool {
1206        self.remote == other.remote
1207            && self.lane() == other.lane()
1208            && self.len().saturating_add(other.len()) <= max_payloads
1209    }
1210
1211    pub(crate) fn remote(&self) -> PeerIdentity {
1212        self.remote
1213    }
1214
1215    pub(crate) fn drop_on_backpressure(&self) -> bool {
1216        self.payloads
1217            .iter()
1218            .all(EndpointDataPayload::drop_on_backpressure)
1219    }
1220
1221    pub(crate) fn into_parts(
1222        self,
1223    ) -> (
1224        PeerIdentity,
1225        Vec<EndpointDataPayload>,
1226        Option<crate::perf_profile::TraceStamp>,
1227    ) {
1228        (self.remote, self.payloads, self.queued_at)
1229    }
1230}
1231
1232impl NodeEndpointCommand {
1233    pub(crate) fn send(
1234        remote: PeerIdentity,
1235        payload: Vec<u8>,
1236        queued_at: Option<crate::perf_profile::TraceStamp>,
1237        response_tx: tokio::sync::oneshot::Sender<Result<(), NodeError>>,
1238    ) -> Self {
1239        Self::Send {
1240            command: EndpointSendCommand::new(remote, payload, queued_at),
1241            response_tx,
1242        }
1243    }
1244
1245    pub(crate) fn send_oneway(
1246        remote: PeerIdentity,
1247        payload: Vec<u8>,
1248        queued_at: Option<crate::perf_profile::TraceStamp>,
1249    ) -> Self {
1250        Self::SendOneway {
1251            command: EndpointSendCommand::new(remote, payload, queued_at),
1252        }
1253    }
1254
1255    pub(crate) fn send_batch_oneway(
1256        remote: PeerIdentity,
1257        payloads: Vec<EndpointDataPayload>,
1258        queued_at: Option<crate::perf_profile::TraceStamp>,
1259        lane: EndpointCommandLane,
1260    ) -> Option<Self> {
1261        debug_assert!(payloads.iter().all(|payload| payload.lane() == lane));
1262        let command = EndpointSendBatchCommand::new(remote, payloads, queued_at)?;
1263        debug_assert_eq!(command.lane(), lane);
1264        Some(Self::SendBatchOneway { command, lane })
1265    }
1266
1267    pub(crate) fn lane(&self) -> EndpointCommandLane {
1268        match self {
1269            Self::Send { command, .. } | Self::SendOneway { command } => command.lane(),
1270            Self::SendBatchOneway { lane, .. } => *lane,
1271            Self::PeerSnapshot { .. }
1272            | Self::RelaySnapshot { .. }
1273            | Self::UpdateRelays { .. }
1274            | Self::UpdatePeers { .. }
1275            | Self::RefreshPeerPaths { .. } => EndpointCommandLane::Priority,
1276        }
1277    }
1278
1279    pub(crate) fn drop_on_backpressure(&self) -> bool {
1280        match self {
1281            Self::SendOneway { command } => {
1282                command.lane() == EndpointCommandLane::Bulk && command.drop_on_backpressure()
1283            }
1284            Self::SendBatchOneway { command, lane } => {
1285                *lane == EndpointCommandLane::Bulk && command.drop_on_backpressure()
1286            }
1287            Self::Send { .. }
1288            | Self::PeerSnapshot { .. }
1289            | Self::RelaySnapshot { .. }
1290            | Self::UpdateRelays { .. }
1291            | Self::UpdatePeers { .. }
1292            | Self::RefreshPeerPaths { .. } => false,
1293        }
1294    }
1295
1296    pub(crate) fn drain_cost(&self) -> usize {
1297        match self {
1298            Self::SendBatchOneway { command, .. } => endpoint_send_batch_drain_cost(command.len()),
1299            Self::Send { .. }
1300            | Self::SendOneway { .. }
1301            | Self::PeerSnapshot { .. }
1302            | Self::RelaySnapshot { .. }
1303            | Self::UpdateRelays { .. }
1304            | Self::UpdatePeers { .. }
1305            | Self::RefreshPeerPaths { .. } => 1,
1306        }
1307    }
1308
1309    pub(crate) fn packet_count(&self) -> usize {
1310        match self {
1311            Self::SendBatchOneway { command, .. } => command.len(),
1312            Self::Send { .. }
1313            | Self::SendOneway { .. }
1314            | Self::PeerSnapshot { .. }
1315            | Self::RelaySnapshot { .. }
1316            | Self::UpdateRelays { .. }
1317            | Self::UpdatePeers { .. }
1318            | Self::RefreshPeerPaths { .. } => 1,
1319        }
1320    }
1321
1322    pub(crate) fn into_send_batch_oneway(
1323        self,
1324    ) -> Result<(EndpointSendBatchCommand, EndpointCommandLane), Self> {
1325        match self {
1326            Self::SendBatchOneway { command, lane } => Ok((command, lane)),
1327            other => Err(other),
1328        }
1329    }
1330}
1331
1332/// Reports what changed in response to `UpdatePeers`.
1333#[derive(Debug, Clone, Default, PartialEq, Eq)]
1334pub(crate) struct UpdatePeersOutcome {
1335    pub(crate) added: usize,
1336    pub(crate) removed: usize,
1337    pub(crate) updated: usize,
1338    pub(crate) unchanged: usize,
1339}
1340
1341/// Authenticated endpoint data emitted by the session receive path.
1342///
1343/// Keeping source identity and payload together makes the delivery-side
1344/// ownership boundary explicit for the current rx loop and for a future
1345/// peer/session runtime that can move endpoint-data delivery off the bounce path.
1346#[derive(Debug)]
1347pub(crate) struct EndpointDataDelivery {
1348    pub(crate) source_peer: PeerIdentity,
1349    pub(crate) payload: PacketBuffer,
1350}
1351
1352impl EndpointDataDelivery {
1353    pub(crate) fn new(source_peer: PeerIdentity, payload: impl Into<PacketBuffer>) -> Self {
1354        Self {
1355            source_peer,
1356            payload: payload.into(),
1357        }
1358    }
1359
1360    fn is_priority_sized(&self) -> bool {
1361        matches!(
1362            endpoint_event_lane_for_len(self.payload.len()),
1363            EndpointEventLane::Priority
1364        )
1365    }
1366}
1367
1368/// Endpoint data events emitted by the node session receive path.
1369#[derive(Debug)]
1370pub(crate) enum NodeEndpointEvent {
1371    Data {
1372        source_peer: PeerIdentity,
1373        payload: PacketBuffer,
1374        queued_at: Option<crate::perf_profile::TraceStamp>,
1375    },
1376    DataBatch {
1377        messages: Vec<EndpointDataDelivery>,
1378        queued_at: Option<crate::perf_profile::TraceStamp>,
1379    },
1380}
1381
1382#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1383pub(in crate::node) struct EndpointEventDequeueCounts {
1384    pub(in crate::node) total: usize,
1385    pub(in crate::node) priority: usize,
1386    pub(in crate::node) bulk: usize,
1387}
1388
1389impl NodeEndpointEvent {
1390    fn message_count(&self) -> usize {
1391        match self {
1392            NodeEndpointEvent::Data { .. } => 1,
1393            NodeEndpointEvent::DataBatch { messages, .. } => messages.len(),
1394        }
1395    }
1396
1397    pub(in crate::node) fn dequeue_counts(&self) -> EndpointEventDequeueCounts {
1398        match self {
1399            NodeEndpointEvent::Data { payload, .. } => {
1400                let priority = usize::from(payload.len() <= ENDPOINT_EVENT_PRIORITY_MAX_LEN);
1401                EndpointEventDequeueCounts {
1402                    total: 1,
1403                    priority,
1404                    bulk: 1 - priority,
1405                }
1406            }
1407            NodeEndpointEvent::DataBatch { messages, .. } => {
1408                let priority = messages
1409                    .iter()
1410                    .filter(|message| message.is_priority_sized())
1411                    .count();
1412                EndpointEventDequeueCounts {
1413                    total: messages.len(),
1414                    priority,
1415                    bulk: messages.len().saturating_sub(priority),
1416                }
1417            }
1418        }
1419    }
1420
1421    fn queued_at(&self) -> Option<crate::perf_profile::TraceStamp> {
1422        match self {
1423            NodeEndpointEvent::Data { queued_at, .. }
1424            | NodeEndpointEvent::DataBatch { queued_at, .. } => *queued_at,
1425        }
1426    }
1427
1428    fn record_dequeue_wait(&self) {
1429        let queued_at = self.queued_at();
1430        if queued_at.is_none() {
1431            return;
1432        }
1433        let counts = self.dequeue_counts();
1434        crate::perf_profile::record_since_split_count(
1435            crate::perf_profile::Stage::EndpointEventWait,
1436            crate::perf_profile::Stage::EndpointPriorityEventWait,
1437            crate::perf_profile::Stage::EndpointBulkEventWait,
1438            queued_at,
1439            counts.total as u64,
1440            counts.priority as u64,
1441            counts.bulk as u64,
1442        );
1443    }
1444
1445    fn from_delivery_messages(
1446        mut messages: Vec<EndpointDataDelivery>,
1447        queued_at: Option<crate::perf_profile::TraceStamp>,
1448    ) -> Option<Self> {
1449        match messages.len() {
1450            0 => None,
1451            1 => {
1452                let message = messages.pop().expect("one endpoint message should exist");
1453                Some(NodeEndpointEvent::Data {
1454                    source_peer: message.source_peer,
1455                    payload: message.payload,
1456                    queued_at,
1457                })
1458            }
1459            _ => Some(NodeEndpointEvent::DataBatch {
1460                messages,
1461                queued_at,
1462            }),
1463        }
1464    }
1465}
1466
1467/// Authenticated peer state exposed to embedded endpoint callers.
1468#[derive(Debug, Clone, PartialEq, Eq)]
1469pub(crate) struct NodeEndpointPeer {
1470    pub(crate) npub: String,
1471    pub(crate) node_addr: NodeAddr,
1472    pub(crate) connected: bool,
1473    pub(crate) transport_addr: Option<String>,
1474    pub(crate) transport_type: Option<String>,
1475    pub(crate) link_id: u64,
1476    pub(crate) srtt_ms: Option<u64>,
1477    pub(crate) srtt_age_ms: Option<u64>,
1478    pub(crate) packets_sent: u64,
1479    pub(crate) packets_recv: u64,
1480    pub(crate) bytes_sent: u64,
1481    pub(crate) bytes_recv: u64,
1482    pub(crate) rekey_in_progress: bool,
1483    pub(crate) rekey_draining: bool,
1484    pub(crate) current_k_bit: Option<bool>,
1485    pub(crate) direct_probe_pending: bool,
1486    pub(crate) direct_probe_after_ms: Option<u64>,
1487    pub(crate) direct_probe_retry_count: u32,
1488    pub(crate) direct_probe_auto_reconnect: bool,
1489    pub(crate) direct_probe_expires_at_ms: Option<u64>,
1490    pub(crate) nostr_traversal_consecutive_failures: u32,
1491    pub(crate) nostr_traversal_in_cooldown: bool,
1492    pub(crate) nostr_traversal_cooldown_until_ms: Option<u64>,
1493    pub(crate) nostr_traversal_last_observed_skew_ms: Option<i64>,
1494}
1495
1496/// Live Nostr relay state exposed to embedded endpoint callers.
1497#[derive(Debug, Clone, PartialEq, Eq)]
1498pub(crate) struct NodeEndpointRelayStatus {
1499    pub(crate) url: String,
1500    pub(crate) status: String,
1501}
1502
1503#[cfg(all(test, unix))]
1504mod endpoint_committed_bulk_tests {
1505    use super::*;
1506    use crate::node::encrypt_worker::DEFAULT_SEND_WEIGHT;
1507    use crate::transport::udp::socket::UdpRawSocket;
1508    use ring::aead::{LessSafeKey, UnboundKey};
1509
1510    #[test]
1511    fn committed_bulk_ready_waits_for_commit_or_cancel() {
1512        let ready = Arc::new(EndpointCommittedBulkReady::new());
1513        assert_eq!(ready.try_state(), EndpointCommittedBulkState::Pending);
1514        let thread_ready = Arc::clone(&ready);
1515        let done = Arc::new(std::sync::atomic::AtomicBool::new(false));
1516        let thread_done = Arc::clone(&done);
1517        let handle = std::thread::spawn(move || {
1518            assert_eq!(thread_ready.wait(), EndpointCommittedBulkState::Committed);
1519            thread_done.store(true, Relaxed);
1520        });
1521
1522        std::thread::sleep(std::time::Duration::from_millis(20));
1523        assert!(
1524            !done.load(Relaxed),
1525            "staged bulk container must stay locked until feedback commits"
1526        );
1527        ready.commit();
1528        handle.join().expect("commit waiter should finish");
1529        assert!(done.load(Relaxed));
1530
1531        let ready = EndpointCommittedBulkReady::new();
1532        ready.cancel();
1533        assert_eq!(ready.try_state(), EndpointCommittedBulkState::Canceled);
1534        assert_eq!(ready.wait(), EndpointCommittedBulkState::Canceled);
1535    }
1536
1537    fn test_cipher() -> LessSafeKey {
1538        let unbound =
1539            UnboundKey::new(&ring::aead::CHACHA20_POLY1305, &[0u8; 32]).expect("build key");
1540        LessSafeKey::new(unbound)
1541    }
1542
1543    fn with_test_socket(
1544        test: impl FnOnce(crate::transport::udp::socket::AsyncUdpSocket, LessSafeKey),
1545    ) {
1546        let rt = tokio::runtime::Builder::new_current_thread()
1547            .enable_io()
1548            .build()
1549            .expect("tokio runtime");
1550        rt.block_on(async {
1551            let raw = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 1 << 20, 1 << 20)
1552                .expect("open UDP socket");
1553            test(raw.into_async().expect("async UDP socket"), test_cipher());
1554        });
1555    }
1556
1557    fn bulk_job(
1558        socket: crate::transport::udp::socket::AsyncUdpSocket,
1559        cipher: &LessSafeKey,
1560        dest_addr: &str,
1561        bulk_endpoint_data: bool,
1562    ) -> crate::node::encrypt_worker::FmpSendJob {
1563        let mut wire_buf = Vec::with_capacity(crate::node::wire::ESTABLISHED_HEADER_SIZE + 32);
1564        wire_buf.extend_from_slice(&[0u8; crate::node::wire::ESTABLISHED_HEADER_SIZE]);
1565        crate::node::encrypt_worker::FmpSendJob {
1566            cipher: cipher.clone(),
1567            counter: 0,
1568            wire_buf,
1569            fsp_seal: None,
1570            send_target: crate::node::encrypt_worker::SelectedSendTarget::new(
1571                socket,
1572                #[cfg(any(target_os = "linux", target_os = "macos"))]
1573                None,
1574                dest_addr.parse().expect("socket addr"),
1575            ),
1576            endpoint_flow_dispatch_key: None,
1577            bulk_endpoint_data,
1578            drop_on_backpressure: bulk_endpoint_data,
1579            scheduling_weight: DEFAULT_SEND_WEIGHT,
1580            queued_at: None,
1581        }
1582    }
1583
1584    #[test]
1585    fn committed_bulk_batch_merge_requires_same_bulk_target() {
1586        with_test_socket(|socket, cipher| {
1587            let workers = crate::node::encrypt_worker::EncryptWorkerPool::spawn(1);
1588            let first = EndpointCommittedBulkBatch {
1589                workers: workers.clone(),
1590                jobs: vec![bulk_job(socket.clone(), &cipher, "127.0.0.1:10031", true)],
1591                ready: Arc::new(EndpointCommittedBulkReady::new()),
1592            };
1593            let same = EndpointCommittedBulkBatch {
1594                workers: workers.clone(),
1595                jobs: vec![bulk_job(socket.clone(), &cipher, "127.0.0.1:10031", true)],
1596                ready: Arc::new(EndpointCommittedBulkReady::new()),
1597            };
1598            let other_target = EndpointCommittedBulkBatch {
1599                workers: workers.clone(),
1600                jobs: vec![bulk_job(socket.clone(), &cipher, "127.0.0.1:10032", true)],
1601                ready: Arc::new(EndpointCommittedBulkReady::new()),
1602            };
1603            let priority_like = EndpointCommittedBulkBatch {
1604                workers,
1605                jobs: vec![bulk_job(socket, &cipher, "127.0.0.1:10031", false)],
1606                ready: Arc::new(EndpointCommittedBulkReady::new()),
1607            };
1608
1609            assert!(first.can_merge_after(&same));
1610            assert!(!first.can_merge_after(&other_target));
1611            assert!(!first.can_merge_after(&priority_like));
1612        });
1613    }
1614}
fips_core/node/endpoint_event.rs

fips_core/node/
endpoint_event.rs
