hopr_transport/
constants.rs

1use std::time::Duration;
2
3/// The maximum waiting time for a message send to produce a half-key challenge reply
4pub const PACKET_QUEUE_TIMEOUT_MILLISECONDS: std::time::Duration = std::time::Duration::from_millis(15000);
5
6/// Maximum number of outgoing application-layer packets buffered before the writer
7/// observes backpressure (`Poll::Pending` on `poll_write`).
8///
9/// Caps the burst submitted to `buffered(8×N_cpus)` so tail packets never
10/// exceed `PACKET_ENCODING_TIMEOUT` (150 ms). With Sphinx encoding at ~21 ms/packet
11/// and 256 slots, the worst-case Rayon queue depth stays well under the timeout on
12/// machines with up to ~36 cores. Safe formula: `≤ 7 × available_parallelism()`.
13pub(crate) const MAXIMUM_MSG_OUTGOING_BUFFER_SIZE: usize = 256;
14
15/// Time within Start protocol must finish session initiation.
16/// This base value is always multiplied by the (max) number of hops, times 2 (for both-ways).
17pub(crate) const SESSION_INITIATION_TIMEOUT_BASE: Duration = Duration::from_secs(5);
18
19#[cfg(test)]
20mod tests {
21    use super::MAXIMUM_MSG_OUTGOING_BUFFER_SIZE;
22
23    /// Guard against accidentally inflating `MAXIMUM_MSG_OUTGOING_BUFFER_SIZE` back to a large
24    /// value that would overflow the Rayon encoding queue.
25    ///
26    /// Safe formula: ≤ 7 × available_parallelism. 256 covers machines with up to ~36 cores.
27    #[test]
28    fn outgoing_buffer_size_within_backpressure_limit() {
29        assert!(
30            MAXIMUM_MSG_OUTGOING_BUFFER_SIZE <= 256,
31            "MAXIMUM_MSG_OUTGOING_BUFFER_SIZE={MAXIMUM_MSG_OUTGOING_BUFFER_SIZE} exceeds the safe threshold; tail \
32             packets will exceed PACKET_ENCODING_TIMEOUT under burst writes"
33        );
34    }
35
36    /// Verify that `CrossfireSink` signals backpressure (`Poll::Pending`) once the channel
37    /// reaches `MAXIMUM_MSG_OUTGOING_BUFFER_SIZE`, preventing the Rayon encoding queue from
38    /// receiving an unbounded burst.
39    #[test]
40    fn outgoing_channel_signals_backpressure_when_full() {
41        use std::{
42            pin::Pin,
43            task::{Context, Poll},
44        };
45
46        use futures::Sink;
47        use hopr_utils::network_types::crossfire_sink::bounded_sink_channel;
48
49        let (mut sink, _rx) = bounded_sink_channel::<usize>(MAXIMUM_MSG_OUTGOING_BUFFER_SIZE);
50        let waker = futures::task::noop_waker_ref();
51        let mut cx = Context::from_waker(waker);
52
53        // Standard Sink protocol: each poll_ready → start_send pair sends one item.
54        // At i=0 the channel is empty; at i=N-1 the final item is buffered.
55        for i in 0..MAXIMUM_MSG_OUTGOING_BUFFER_SIZE {
56            assert!(
57                matches!(Pin::new(&mut sink).poll_ready(&mut cx), Poll::Ready(Ok(()))),
58                "poll_ready must be Ready before capacity is reached (item {i})"
59            );
60            Pin::new(&mut sink).start_send(i).unwrap();
61        }
62        // This poll_ready sends the last buffered item; channel is now exactly full.
63        assert!(
64            matches!(Pin::new(&mut sink).poll_ready(&mut cx), Poll::Ready(Ok(()))),
65            "poll_ready must be Ready when flushing the final item into a full-but-not-yet-full channel"
66        );
67
68        // One extra item: buffer it, then poll_ready must indicate the channel is saturated.
69        Pin::new(&mut sink)
70            .start_send(MAXIMUM_MSG_OUTGOING_BUFFER_SIZE)
71            .unwrap();
72        assert!(
73            matches!(Pin::new(&mut sink).poll_ready(&mut cx), Poll::Pending),
74            "CrossfireSink must return Poll::Pending when channel is at capacity ({})",
75            MAXIMUM_MSG_OUTGOING_BUFFER_SIZE
76        );
77    }
78}
hopr_transport/constants.rs

hopr_transport/
constants.rs
