Skip to main content

net_io/
limits.rs

1//! Every bound net-io enforces, in one value.
2
3use std::time::Duration;
4
5use crate::addr::PrefixPolicy;
6use crate::driver::DriverKind;
7use crate::error::{ConfigReason, NetError};
8use crate::info::Lane;
9
10/// Lanes that can hold a connection reservation (0 to 7).
11const RESERVED_LANES: usize = 8;
12
13/// A minimum transfer rate, checked per direction while that direction is active.
14///
15/// # Examples
16///
17/// ```
18/// use std::time::Duration;
19/// let rate = net_io::MinRate::new(512, Duration::from_secs(10));
20/// assert_eq!(rate.grace, Duration::from_secs(5));
21/// assert!(net_io::MinRate::off().is_off());
22/// ```
23#[derive(Clone, Copy, Debug, PartialEq, Eq)]
24pub struct MinRate {
25    /// Bytes per second the peer must reach over the window.
26    pub bytes_per_sec: u32,
27    /// The sliding window.
28    pub window: Duration,
29    /// How long after a direction becomes active the check starts.
30    pub grace: Duration,
31    /// How often the rate is checked.
32    pub check: Duration,
33}
34
35impl MinRate {
36    /// No minimum rate.
37    #[must_use]
38    pub const fn off() -> MinRate {
39        MinRate {
40            bytes_per_sec: 0,
41            window: Duration::from_secs(10),
42            grace: Duration::from_secs(5),
43            check: Duration::from_secs(1),
44        }
45    }
46
47    /// `bytes_per_sec` over `window`, after a 5 s grace, checked every second.
48    #[must_use]
49    pub const fn new(bytes_per_sec: u32, window: Duration) -> MinRate {
50        MinRate {
51            bytes_per_sec,
52            window,
53            grace: Duration::from_secs(5),
54            check: Duration::from_secs(1),
55        }
56    }
57
58    /// Whether the rate check is off.
59    #[must_use]
60    pub const fn is_off(&self) -> bool {
61        self.bytes_per_sec == 0
62    }
63}
64
65/// Adaptive deadlines under pressure (applied from `0.6.0`).
66///
67/// ```
68/// let p = net_io::Pressure::default();
69/// assert!(p.high > p.low);
70/// ```
71#[derive(Clone, Copy, Debug, PartialEq)]
72pub struct Pressure {
73    /// Fraction of `max_connections` above which deadlines shrink.
74    pub high: f32,
75    /// Fraction below which they return to normal.
76    pub low: f32,
77    /// The multiplier applied to handshake and idle timeouts.
78    pub factor: f32,
79}
80
81impl Default for Pressure {
82    fn default() -> Pressure {
83        Pressure {
84            high: 0.8,
85            low: 0.6,
86            factor: 0.25,
87        }
88    }
89}
90
91/// A buffer size class: a power of two from 1 KiB to 1 MiB.
92///
93/// ```
94/// use net_io::SizeClass;
95/// assert_eq!(SizeClass::MEDIUM.bytes(), 16 * 1024);
96/// assert!(SizeClass::new(3000).is_err());
97/// assert_eq!(SizeClass::new(2048).unwrap().bytes(), 2048);
98/// ```
99#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
100pub struct SizeClass(u32);
101
102impl SizeClass {
103    /// 4 KiB.
104    pub const SMALL: SizeClass = SizeClass(4 * 1024);
105    /// 16 KiB.
106    pub const MEDIUM: SizeClass = SizeClass(16 * 1024);
107    /// 64 KiB.
108    pub const LARGE: SizeClass = SizeClass(64 * 1024);
109
110    /// A class of `bytes`.
111    ///
112    /// # Errors
113    ///
114    /// `InvalidConfig { field: "size_class" }` unless `bytes` is a power of two from 1 KiB to 1 MiB.
115    pub const fn new(bytes: u32) -> Result<SizeClass, NetError> {
116        if !bytes.is_power_of_two() {
117            return Err(NetError::InvalidConfig {
118                field: "size_class",
119                reason: ConfigReason::Inconsistent,
120            });
121        }
122        if bytes < 1024 {
123            return Err(NetError::InvalidConfig {
124                field: "size_class",
125                reason: ConfigReason::TooSmall { min: 1024 },
126            });
127        }
128        if bytes > 1024 * 1024 {
129            return Err(NetError::InvalidConfig {
130                field: "size_class",
131                reason: ConfigReason::TooLarge { max: 1024 * 1024 },
132            });
133        }
134        Ok(SizeClass(bytes))
135    }
136
137    /// The size in bytes.
138    #[must_use]
139    pub const fn bytes(self) -> u32 {
140        self.0
141    }
142}
143
144/// Every bound net-io enforces. Unset values take the defaults of the driver
145/// that runs the connection.
146///
147/// # Examples
148///
149/// ```
150/// use net_io::{DriverKind, Limits, MinRate};
151/// use std::time::Duration;
152///
153/// let limits = Limits::builder()
154///     .max_connections(20_000)
155///     .idle_timeout(Duration::from_secs(15))
156///     .min_rate(MinRate::new(512, Duration::from_secs(10)))
157///     .build()?;
158/// assert_eq!(limits.max_connections(), Some(20_000));
159/// assert_eq!(limits.read_timeout(), None);
160/// assert_eq!(limits.resolve(DriverKind::Blocking).read_timeout(), Some(Duration::from_secs(30)));
161/// # Ok::<(), net_io::NetError>(())
162/// ```
163#[non_exhaustive]
164#[derive(Clone, Debug, Default, PartialEq)]
165pub struct Limits {
166    max_connections: Option<u32>,
167    max_per_listener: Option<u32>,
168    max_per_source: Option<(u32, PrefixPolicy)>,
169    backlog: Option<u32>,
170    accept_batch: Option<u16>,
171    handshake_timeout: Option<Duration>,
172    idle_timeout: Option<Duration>,
173    read_timeout: Option<Duration>,
174    write_timeout: Option<Duration>,
175    park_timeout: Option<Duration>,
176    min_rate: Option<MinRate>,
177    proxy_header_max: Option<u16>,
178    read_class: Option<(SizeClass, SizeClass)>,
179    output_budget: Option<u32>,
180    completion_queue: Option<u32>,
181    buffer_budget: Option<u64>,
182    drain_timeout: Option<Duration>,
183    graceful_linger: Option<(Duration, u32)>,
184    refusal_bytes_max: Option<u16>,
185    tick: Option<Duration>,
186    pressure: Option<Pressure>,
187    lane_reserve: [Option<u32>; RESERVED_LANES],
188}
189
190/// The resolved values a running connection uses.
191#[derive(Clone, Copy, Debug, PartialEq)]
192pub(crate) struct Effective {
193    pub max_connections: u32,
194    pub max_per_listener: u32,
195    pub max_per_source: u32,
196    pub source_policy: PrefixPolicy,
197    pub backlog: u32,
198    pub accept_batch: u16,
199    pub handshake_timeout: Duration,
200    pub idle_timeout: Duration,
201    pub read_timeout: Duration,
202    pub write_timeout: Duration,
203    pub park_timeout: Duration,
204    pub min_rate: MinRate,
205    pub proxy_header_max: u16,
206    pub read_class: SizeClass,
207    pub read_class_max: SizeClass,
208    pub output_budget: u32,
209    pub completion_queue: u32,
210    pub buffer_budget: u64,
211    pub drain_timeout: Duration,
212    pub graceful_linger: Duration,
213    pub graceful_linger_bytes: u32,
214    pub refusal_bytes_max: u16,
215    pub tick: Duration,
216}
217
218impl Effective {
219    /// The defaults of `driver` (blocking, or the reactor).
220    pub(crate) fn defaults(driver: DriverKind) -> Effective {
221        let blocking = driver.resolve() == DriverKind::Blocking;
222        let max_connections = if blocking { 128 } else { 10_000 };
223        Effective {
224            max_connections,
225            max_per_listener: max_connections,
226            max_per_source: if blocking { 16 } else { 256 },
227            source_policy: PrefixPolicy::DEFAULT,
228            backlog: 1024,
229            accept_batch: if blocking { 16 } else { 64 },
230            handshake_timeout: Duration::from_secs(10),
231            idle_timeout: Duration::from_secs(if blocking { 5 } else { 60 }),
232            read_timeout: Duration::from_secs(30),
233            write_timeout: Duration::from_secs(30),
234            park_timeout: Duration::from_secs(30),
235            min_rate: MinRate::new(256, Duration::from_secs(10)),
236            proxy_header_max: 1024,
237            read_class: SizeClass::MEDIUM,
238            read_class_max: SizeClass::LARGE,
239            output_budget: 64 * 1024,
240            completion_queue: if blocking { 1 } else { 1024 },
241            buffer_budget: if blocking { 64 << 20 } else { 512 << 20 },
242            drain_timeout: Duration::from_secs(30),
243            graceful_linger: Duration::from_secs(2),
244            graceful_linger_bytes: 64 * 1024,
245            refusal_bytes_max: 512,
246            tick: Duration::from_millis(if blocking { 10 } else { 1 }),
247        }
248    }
249}
250
251impl Limits {
252    /// A builder with every value unset.
253    ///
254    /// ```
255    /// let limits = net_io::Limits::builder().max_connections(100).build()?;
256    /// assert_eq!(limits.max_per_source(), None);
257    /// # Ok::<(), net_io::NetError>(())
258    /// ```
259    pub fn builder() -> LimitsBuilder {
260        LimitsBuilder::default()
261    }
262
263    /// Every value set to the reactor's defaults.
264    ///
265    /// ```
266    /// assert_eq!(net_io::Limits::reactor_defaults().max_connections(), Some(10_000));
267    /// ```
268    #[must_use]
269    pub fn reactor_defaults() -> Limits {
270        Limits::from_effective(&Effective::defaults(DriverKind::Reactor))
271    }
272
273    /// Every value set to the blocking driver's defaults.
274    ///
275    /// ```
276    /// assert_eq!(net_io::Limits::blocking_defaults().idle_timeout(), Some(std::time::Duration::from_secs(5)));
277    /// ```
278    #[must_use]
279    pub fn blocking_defaults() -> Limits {
280        Limits::from_effective(&Effective::defaults(DriverKind::Blocking))
281    }
282
283    /// A builder starting from these values.
284    ///
285    /// Every value carried over counts as set, so lowering `max_connections`
286    /// below a carried-over `max_per_listener` fails validation. Start from
287    /// [`Limits::builder`] to let dependent values follow `max_connections`.
288    ///
289    /// ```
290    /// let limits = net_io::Limits::blocking_defaults().to_builder().max_connections(256).build()?;
291    /// assert_eq!(limits.max_connections(), Some(256));
292    /// # Ok::<(), net_io::NetError>(())
293    /// ```
294    pub fn to_builder(&self) -> LimitsBuilder {
295        LimitsBuilder {
296            limits: self.clone(),
297        }
298    }
299
300    /// These limits with unset values filled from `driver`'s defaults.
301    ///
302    /// ```
303    /// let limits = net_io::Limits::builder().max_connections(10).build()?;
304    /// let resolved = limits.resolve(net_io::DriverKind::Blocking);
305    /// assert_eq!(resolved.max_per_source().map(|(n, _)| n), Some(10));
306    /// # Ok::<(), net_io::NetError>(())
307    /// ```
308    #[must_use]
309    pub fn resolve(&self, driver: DriverKind) -> Limits {
310        let mut out = Limits::from_effective(&self.effective(driver));
311        out.pressure = self.pressure;
312        out.lane_reserve = self.lane_reserve;
313        out
314    }
315
316    /// Checks the resolved values against each other, the worker count, and the
317    /// process's descriptor limit.
318    ///
319    /// # Errors
320    ///
321    /// `InvalidConfig { field, reason }` for the first rule broken.
322    ///
323    /// ```
324    /// let limits = net_io::Limits::builder().max_connections(1_000_000).build()?;
325    /// assert!(limits.validate(4, Some(1024)).is_err());
326    /// # Ok::<(), net_io::NetError>(())
327    /// ```
328    pub fn validate(&self, workers: u16, descriptor_limit: Option<u64>) -> Result<(), NetError> {
329        // Rules that need only the values themselves.
330        LimitsBuilder {
331            limits: self.clone(),
332        }
333        .check()?;
334        let e = self.effective(DriverKind::Blocking);
335        let need_budget = u64::from(workers)
336            .saturating_mul(u64::from(e.read_class_max.bytes()))
337            .saturating_mul(2);
338        if e.buffer_budget < need_budget {
339            return Err(invalid(
340                "buffer_budget",
341                ConfigReason::BudgetBelowWorkers { need: need_budget },
342            ));
343        }
344        if let Some(have) = descriptor_limit {
345            let need = descriptors_needed(e.max_connections, workers);
346            if need > have {
347                return Err(invalid(
348                    "max_connections",
349                    ConfigReason::DescriptorLimit { have, need },
350                ));
351            }
352        }
353        Ok(())
354    }
355
356    /// The resolved values for `driver`.
357    pub(crate) fn effective(&self, driver: DriverKind) -> Effective {
358        let d = Effective::defaults(driver);
359        let max_connections = self.max_connections.unwrap_or(d.max_connections);
360        let (max_per_source, source_policy) = self
361            .max_per_source
362            .unwrap_or((d.max_per_source.min(max_connections), d.source_policy));
363        let (read_class, read_class_max) =
364            self.read_class.unwrap_or((d.read_class, d.read_class_max));
365        let (graceful_linger, graceful_linger_bytes) = self
366            .graceful_linger
367            .unwrap_or((d.graceful_linger, d.graceful_linger_bytes));
368        Effective {
369            max_connections,
370            max_per_listener: self.max_per_listener.unwrap_or(max_connections),
371            max_per_source,
372            source_policy,
373            backlog: self.backlog.unwrap_or(d.backlog),
374            accept_batch: self.accept_batch.unwrap_or(d.accept_batch),
375            handshake_timeout: self.handshake_timeout.unwrap_or(d.handshake_timeout),
376            idle_timeout: self.idle_timeout.unwrap_or(d.idle_timeout),
377            read_timeout: self.read_timeout.unwrap_or(d.read_timeout),
378            write_timeout: self.write_timeout.unwrap_or(d.write_timeout),
379            park_timeout: self.park_timeout.unwrap_or(d.park_timeout),
380            min_rate: self.min_rate.unwrap_or(d.min_rate),
381            proxy_header_max: self.proxy_header_max.unwrap_or(d.proxy_header_max),
382            read_class,
383            read_class_max,
384            output_budget: self.output_budget.unwrap_or(d.output_budget),
385            completion_queue: self.completion_queue.unwrap_or(d.completion_queue),
386            buffer_budget: self.buffer_budget.unwrap_or(d.buffer_budget),
387            drain_timeout: self.drain_timeout.unwrap_or(d.drain_timeout),
388            graceful_linger,
389            graceful_linger_bytes,
390            refusal_bytes_max: self.refusal_bytes_max.unwrap_or(d.refusal_bytes_max),
391            tick: self.tick.unwrap_or(d.tick),
392        }
393    }
394
395    /// Values from `other` where it sets them, otherwise from `self` (listener overrides).
396    pub(crate) fn overlay(&self, other: &Limits) -> Limits {
397        macro_rules! pick {
398            ($($f:ident),*) => { Limits { $($f: other.$f.or(self.$f),)* lane_reserve: {
399                let mut r = self.lane_reserve;
400                for (i, v) in other.lane_reserve.iter().enumerate() { if v.is_some() { r[i] = *v; } }
401                r
402            } } };
403        }
404        pick!(
405            max_connections,
406            max_per_listener,
407            max_per_source,
408            backlog,
409            accept_batch,
410            handshake_timeout,
411            idle_timeout,
412            read_timeout,
413            write_timeout,
414            park_timeout,
415            min_rate,
416            proxy_header_max,
417            read_class,
418            output_budget,
419            completion_queue,
420            buffer_budget,
421            drain_timeout,
422            graceful_linger,
423            refusal_bytes_max,
424            tick,
425            pressure
426        )
427    }
428
429    fn from_effective(e: &Effective) -> Limits {
430        Limits {
431            max_connections: Some(e.max_connections),
432            max_per_listener: Some(e.max_per_listener),
433            max_per_source: Some((e.max_per_source, e.source_policy)),
434            backlog: Some(e.backlog),
435            accept_batch: Some(e.accept_batch),
436            handshake_timeout: Some(e.handshake_timeout),
437            idle_timeout: Some(e.idle_timeout),
438            read_timeout: Some(e.read_timeout),
439            write_timeout: Some(e.write_timeout),
440            park_timeout: Some(e.park_timeout),
441            min_rate: Some(e.min_rate),
442            proxy_header_max: Some(e.proxy_header_max),
443            read_class: Some((e.read_class, e.read_class_max)),
444            output_budget: Some(e.output_budget),
445            completion_queue: Some(e.completion_queue),
446            buffer_budget: Some(e.buffer_budget),
447            drain_timeout: Some(e.drain_timeout),
448            graceful_linger: Some((e.graceful_linger, e.graceful_linger_bytes)),
449            refusal_bytes_max: Some(e.refusal_bytes_max),
450            tick: Some(e.tick),
451            pressure: None,
452            lane_reserve: [None; RESERVED_LANES],
453        }
454    }
455
456    /// The global connection cap.
457    #[must_use]
458    pub fn max_connections(&self) -> Option<u32> {
459        self.max_connections
460    }
461    /// The per-listener connection cap.
462    #[must_use]
463    pub fn max_per_listener(&self) -> Option<u32> {
464        self.max_per_listener
465    }
466    /// The per-source concurrency cap and how sources are grouped. Each listener
467    /// applies its own value to its own connections. Every open connection counts,
468    /// including one finishing a graceful close, so a client at the cap that
469    /// reconnects at once can be refused until its last connection is closed.
470    #[must_use]
471    pub fn max_per_source(&self) -> Option<(u32, PrefixPolicy)> {
472        self.max_per_source
473    }
474    /// The listen backlog.
475    #[must_use]
476    pub fn backlog(&self) -> Option<u32> {
477        self.backlog
478    }
479    /// Connections accepted per wake-up.
480    #[must_use]
481    pub fn accept_batch(&self) -> Option<u16> {
482        self.accept_batch
483    }
484    /// Time for the PROXY header and TLS handshake together.
485    #[must_use]
486    pub fn handshake_timeout(&self) -> Option<Duration> {
487        self.handshake_timeout
488    }
489    /// Time to wait for the next exchange.
490    #[must_use]
491    pub fn idle_timeout(&self) -> Option<Duration> {
492        self.idle_timeout
493    }
494    /// Time to wait for bytes during an exchange.
495    #[must_use]
496    pub fn read_timeout(&self) -> Option<Duration> {
497        self.read_timeout
498    }
499    /// Time to wait for the peer to take bytes.
500    #[must_use]
501    pub fn write_timeout(&self) -> Option<Duration> {
502        self.write_timeout
503    }
504    /// Time a parked connection may wait for its completion.
505    #[must_use]
506    pub fn park_timeout(&self) -> Option<Duration> {
507        self.park_timeout
508    }
509    /// The minimum transfer rate.
510    #[must_use]
511    pub fn min_rate(&self) -> Option<MinRate> {
512        self.min_rate
513    }
514    /// The PROXY v2 address-and-TLV cap.
515    #[must_use]
516    pub fn proxy_header_max(&self) -> Option<u16> {
517        self.proxy_header_max
518    }
519    /// The initial and maximum read buffer class.
520    #[must_use]
521    pub fn read_class(&self) -> Option<(SizeClass, SizeClass)> {
522        self.read_class
523    }
524    /// Pool memory a connection's output may hold.
525    #[must_use]
526    pub fn output_budget(&self) -> Option<u32> {
527        self.output_budget
528    }
529    /// A worker's completion queue length.
530    #[must_use]
531    pub fn completion_queue(&self) -> Option<u32> {
532        self.completion_queue
533    }
534    /// Total bytes of buffers all workers may allocate.
535    #[must_use]
536    pub fn buffer_budget(&self) -> Option<u64> {
537        self.buffer_budget
538    }
539    /// The drain deadline.
540    #[must_use]
541    pub fn drain_timeout(&self) -> Option<Duration> {
542        self.drain_timeout
543    }
544    /// The graceful-close linger: time and bytes read and discarded.
545    #[must_use]
546    pub fn graceful_linger(&self) -> Option<(Duration, u32)> {
547        self.graceful_linger
548    }
549    /// The largest refusal reply.
550    #[must_use]
551    pub fn refusal_bytes_max(&self) -> Option<u16> {
552        self.refusal_bytes_max
553    }
554    /// The timer resolution.
555    #[must_use]
556    pub fn tick(&self) -> Option<Duration> {
557        self.tick
558    }
559    /// Adaptive deadlines under pressure.
560    #[must_use]
561    pub fn pressure(&self) -> Option<Pressure> {
562        self.pressure
563    }
564    /// The connection slots reserved for `lane` (lanes 0 to 7).
565    #[must_use]
566    pub fn lane_reserve(&self, lane: Lane) -> Option<u32> {
567        self.lane_reserve
568            .get(usize::from(lane.0))
569            .copied()
570            .flatten()
571    }
572}
573
574/// Descriptors a configuration needs: connections, two wake descriptors per
575/// worker, and a fixed reserve for listeners and the spare descriptor.
576pub(crate) fn descriptors_needed(max_connections: u32, workers: u16) -> u64 {
577    u64::from(max_connections) + 2 * u64::from(workers) + 32
578}
579
580fn invalid(field: &'static str, reason: ConfigReason) -> NetError {
581    NetError::InvalidConfig { field, reason }
582}
583
584/// Builds [`Limits`]; every setter is optional.
585///
586/// ```
587/// use net_io::{ConfigReason, Limits, NetError};
588/// let err = Limits::builder().max_connections(100).max_per_listener(200).build().unwrap_err();
589/// assert!(matches!(err, NetError::InvalidConfig { field: "max_per_listener", reason: ConfigReason::AboveGlobal }));
590/// assert!(Limits::builder().max_connections(100).build().is_ok());
591/// ```
592#[must_use]
593#[derive(Clone, Debug, Default)]
594pub struct LimitsBuilder {
595    limits: Limits,
596}
597
598macro_rules! setter {
599    ($(#[$m:meta])* $name:ident, $t:ty) => {
600        $(#[$m])*
601        pub fn $name(mut self, value: $t) -> Self {
602            self.limits.$name = Some(value);
603            self
604        }
605    };
606}
607
608impl LimitsBuilder {
609    setter!(/// The global connection cap.
610        max_connections, u32);
611    setter!(/// The per-listener connection cap.
612        max_per_listener, u32);
613    setter!(/// The listen backlog.
614        backlog, u32);
615    setter!(/// Connections accepted per wake-up.
616        accept_batch, u16);
617    setter!(/// Time for the PROXY header and TLS handshake together.
618        handshake_timeout, Duration);
619    setter!(/// Time to wait for the next exchange.
620        idle_timeout, Duration);
621    setter!(/// Time to wait for bytes during an exchange.
622        read_timeout, Duration);
623    setter!(/// Time to wait for the peer to take bytes.
624        write_timeout, Duration);
625    setter!(/// Time a parked connection may wait for its completion.
626        park_timeout, Duration);
627    setter!(/// The minimum transfer rate ([`MinRate::off`] to disable).
628        min_rate, MinRate);
629    setter!(/// Pool memory a connection's output may hold.
630        output_budget, u32);
631    setter!(/// Total bytes of buffers all workers may allocate.
632        buffer_budget, u64);
633    setter!(/// The drain deadline.
634        drain_timeout, Duration);
635    setter!(/// The largest refusal reply (at most 512).
636        refusal_bytes_max, u16);
637    setter!(/// The timer resolution (1 ms to 100 ms).
638        tick, Duration);
639    setter!(/// Adaptive deadlines under pressure (applied from `0.6.0`).
640        pressure, Pressure);
641
642    /// The per-source concurrency cap and how sources are grouped.
643    pub fn max_per_source(mut self, n: u32, policy: PrefixPolicy) -> Self {
644        self.limits.max_per_source = Some((n, policy));
645        self
646    }
647
648    /// The PROXY v2 address-and-TLV cap (up to 65,519).
649    pub fn proxy_header_max(mut self, v2_body_bytes: u16) -> Self {
650        self.limits.proxy_header_max = Some(v2_body_bytes);
651        self
652    }
653
654    /// The initial read buffer class and the largest a session may grow it to.
655    pub fn read_class(mut self, initial: SizeClass, max: SizeClass) -> Self {
656        self.limits.read_class = Some((initial, max));
657        self
658    }
659
660    /// A worker's completion queue length, rounded up to a power of two.
661    pub fn completion_queue(mut self, n: u32) -> Self {
662        self.limits.completion_queue = Some(n);
663        self
664    }
665
666    /// The graceful-close linger: time and bytes read and discarded.
667    pub fn graceful_linger(mut self, time: Duration, bytes: u32) -> Self {
668        self.limits.graceful_linger = Some((time, bytes));
669        self
670    }
671
672    /// Connection slots reserved for `lane` (lanes 0 to 7; applied from `0.6.0`).
673    pub fn lane_reserve(mut self, lane: Lane, slots: u32) -> Self {
674        match self.limits.lane_reserve.get_mut(usize::from(lane.0)) {
675            Some(slot) => *slot = Some(slots),
676            // Rejected by `build` (field "lane_reserve").
677            None => self.limits.lane_reserve[RESERVED_LANES - 1] = Some(u32::MAX),
678        }
679        self
680    }
681
682    /// Checks the values that were set and returns the limits.
683    ///
684    /// # Errors
685    ///
686    /// `InvalidConfig { field, reason }` for the first rule broken, in the order
687    /// documented in `docs/API.md`.
688    pub fn build(self) -> Result<Limits, NetError> {
689        self.check()?;
690        let mut limits = self.limits;
691        if let Some(n) = limits.completion_queue {
692            limits.completion_queue = Some(n.next_power_of_two());
693        }
694        Ok(limits)
695    }
696
697    fn check(&self) -> Result<(), NetError> {
698        let l = &self.limits;
699        // Zero where zero is meaningless.
700        let zeros: [(&'static str, bool); 6] = [
701            ("max_connections", l.max_connections == Some(0)),
702            ("backlog", l.backlog == Some(0)),
703            ("accept_batch", l.accept_batch == Some(0)),
704            ("max_per_source", matches!(l.max_per_source, Some((0, _)))),
705            ("completion_queue", l.completion_queue == Some(0)),
706            ("output_budget", l.output_budget == Some(0)),
707        ];
708        for (field, zero) in zeros {
709            if zero {
710                return Err(invalid(field, ConfigReason::Zero));
711            }
712        }
713        if let Some(max) = l.max_connections {
714            if l.max_per_listener.is_some_and(|n| n > max) {
715                return Err(invalid("max_per_listener", ConfigReason::AboveGlobal));
716            }
717            if l.max_per_source.is_some_and(|(n, _)| n > max) {
718                return Err(invalid("max_per_source", ConfigReason::AboveGlobal));
719            }
720        }
721        if let Some(tick) = l.tick {
722            if tick < Duration::from_millis(1) {
723                return Err(invalid("tick", ConfigReason::TooSmall { min: 1 }));
724            }
725            if tick > Duration::from_millis(100) {
726                return Err(invalid("tick", ConfigReason::TooLarge { max: 100 }));
727            }
728        }
729        let tick = l.tick.unwrap_or(Duration::from_millis(1));
730        let timeouts: [(&'static str, Option<Duration>); 6] = [
731            ("handshake_timeout", l.handshake_timeout),
732            ("idle_timeout", l.idle_timeout),
733            ("read_timeout", l.read_timeout),
734            ("write_timeout", l.write_timeout),
735            ("park_timeout", l.park_timeout),
736            ("drain_timeout", l.drain_timeout),
737        ];
738        for (field, value) in timeouts {
739            if value.is_some_and(|d| d < tick) {
740                return Err(invalid(field, ConfigReason::BelowTick));
741            }
742        }
743        if let Some(rate) = l.min_rate {
744            if !rate.is_off() {
745                if rate.check.is_zero() || rate.grace.is_zero() {
746                    return Err(invalid("min_rate", ConfigReason::Zero));
747                }
748                if rate.window < rate.check {
749                    return Err(invalid("min_rate", ConfigReason::WindowBelowCheck));
750                }
751            }
752        }
753        if let Some((initial, max)) = l.read_class {
754            if max < initial {
755                return Err(invalid("read_class", ConfigReason::Inconsistent));
756            }
757        }
758        if l.refusal_bytes_max.is_some_and(|n| n > 512) {
759            return Err(invalid(
760                "refusal_bytes_max",
761                ConfigReason::TooLarge { max: 512 },
762            ));
763        }
764        if l.proxy_header_max.is_some_and(|n| n > 65_519) {
765            return Err(invalid(
766                "proxy_header_max",
767                ConfigReason::TooLarge { max: 65_519 },
768            ));
769        }
770        if l.completion_queue.is_some_and(|n| n > 1 << 20) {
771            return Err(invalid(
772                "completion_queue",
773                ConfigReason::TooLarge { max: 1 << 20 },
774            ));
775        }
776        if let Some(p) = l.pressure {
777            let factor_ok = p.factor > 0.0 && p.factor <= 1.0;
778            let ordered = p.high.partial_cmp(&p.low) == Some(std::cmp::Ordering::Greater);
779            if !ordered
780                || !factor_ok
781                || !(0.0..=1.0).contains(&p.high)
782                || !(0.0..=1.0).contains(&p.low)
783            {
784                return Err(invalid("pressure", ConfigReason::Inconsistent));
785            }
786        }
787        if l.lane_reserve[RESERVED_LANES - 1] == Some(u32::MAX) {
788            return Err(invalid("lane_reserve", ConfigReason::TooLarge { max: 7 }));
789        }
790        let reserved: u64 = l.lane_reserve.iter().flatten().map(|n| u64::from(*n)).sum();
791        if let Some(max) = l.max_connections {
792            if reserved > u64::from(max) {
793                return Err(invalid("lane_reserve", ConfigReason::AboveGlobal));
794            }
795        }
796        Ok(())
797    }
798}
799
800#[cfg(test)]
801mod tests {
802    use super::*;
803
804    #[test]
805    fn test_unset_values_take_driver_defaults() {
806        let l = Limits::builder().build().unwrap();
807        assert_eq!(
808            l.effective(DriverKind::Blocking).idle_timeout,
809            Duration::from_secs(5)
810        );
811        assert_eq!(
812            l.effective(DriverKind::Reactor).idle_timeout,
813            Duration::from_secs(60)
814        );
815    }
816
817    #[test]
818    fn test_dependent_defaults_follow_max_connections() {
819        let l = Limits::builder().max_connections(10).build().unwrap();
820        let e = l.effective(DriverKind::Reactor);
821        assert_eq!(e.max_per_listener, 10);
822        assert_eq!(e.max_per_source, 10);
823    }
824
825    #[test]
826    fn test_zero_values_rejected() {
827        for b in [
828            Limits::builder().max_connections(0),
829            Limits::builder().backlog(0),
830            Limits::builder().accept_batch(0),
831            Limits::builder().max_per_source(0, PrefixPolicy::DEFAULT),
832            Limits::builder().completion_queue(0),
833            Limits::builder().output_budget(0),
834        ] {
835            assert!(matches!(
836                b.build(),
837                Err(NetError::InvalidConfig {
838                    reason: ConfigReason::Zero,
839                    ..
840                })
841            ));
842        }
843    }
844
845    #[test]
846    fn test_timeout_below_tick_rejected() {
847        let err = Limits::builder()
848            .tick(Duration::from_millis(100))
849            .read_timeout(Duration::from_millis(50))
850            .build()
851            .unwrap_err();
852        assert!(matches!(
853            err,
854            NetError::InvalidConfig {
855                field: "read_timeout",
856                reason: ConfigReason::BelowTick
857            }
858        ));
859    }
860
861    #[test]
862    fn test_min_rate_window_below_check_rejected() {
863        let rate = MinRate {
864            bytes_per_sec: 10,
865            window: Duration::from_millis(500),
866            grace: Duration::from_secs(1),
867            check: Duration::from_secs(1),
868        };
869        assert!(matches!(
870            Limits::builder().min_rate(rate).build(),
871            Err(NetError::InvalidConfig {
872                reason: ConfigReason::WindowBelowCheck,
873                ..
874            })
875        ));
876        assert!(Limits::builder().min_rate(MinRate::off()).build().is_ok());
877    }
878
879    #[test]
880    fn test_bounds_too_large() {
881        assert!(Limits::builder().refusal_bytes_max(513).build().is_err());
882        assert!(Limits::builder().refusal_bytes_max(512).build().is_ok());
883        assert!(Limits::builder().proxy_header_max(65_520).build().is_err());
884        assert!(
885            Limits::builder()
886                .completion_queue((1 << 20) + 1)
887                .build()
888                .is_err()
889        );
890    }
891
892    #[test]
893    fn test_completion_queue_rounded_to_power_of_two() {
894        let l = Limits::builder().completion_queue(1000).build().unwrap();
895        assert_eq!(l.completion_queue(), Some(1024));
896    }
897
898    #[test]
899    fn test_pressure_and_lane_reserve_validation() {
900        let bad = Pressure {
901            high: 0.5,
902            low: 0.6,
903            factor: 0.5,
904        };
905        assert!(Limits::builder().pressure(bad).build().is_err());
906        assert!(Limits::builder().lane_reserve(Lane(8), 1).build().is_err());
907        assert!(
908            Limits::builder()
909                .max_connections(10)
910                .lane_reserve(Lane(0), 6)
911                .lane_reserve(Lane(1), 5)
912                .build()
913                .is_err()
914        );
915    }
916
917    #[test]
918    fn test_validate_descriptor_limit_and_budget() {
919        let l = Limits::builder().max_connections(1000).build().unwrap();
920        assert!(matches!(
921            l.validate(4, Some(500)),
922            Err(NetError::InvalidConfig {
923                reason: ConfigReason::DescriptorLimit { have: 500, .. },
924                ..
925            })
926        ));
927        assert!(l.validate(4, Some(100_000)).is_ok());
928        let small = Limits::builder().buffer_budget(1024).build().unwrap();
929        assert!(matches!(
930            small.validate(4, None),
931            Err(NetError::InvalidConfig {
932                reason: ConfigReason::BudgetBelowWorkers { .. },
933                ..
934            })
935        ));
936    }
937
938    #[test]
939    fn test_overlay_takes_listener_values() {
940        let base = Limits::builder()
941            .max_connections(100)
942            .idle_timeout(Duration::from_secs(9))
943            .build()
944            .unwrap();
945        let over = Limits::builder()
946            .idle_timeout(Duration::from_secs(1))
947            .build()
948            .unwrap();
949        let merged = base.overlay(&over);
950        assert_eq!(merged.max_connections(), Some(100));
951        assert_eq!(merged.idle_timeout(), Some(Duration::from_secs(1)));
952    }
953
954    #[test]
955    fn test_resolve_sets_every_value() {
956        let r = Limits::default().resolve(DriverKind::Blocking);
957        assert!(
958            r.max_connections().is_some() && r.tick().is_some() && r.graceful_linger().is_some()
959        );
960    }
961}