relay_knowledge/net/qos/
mod.rs

1//! QoS admission policy for inbound and outbound network work.
2//!
3//! The policy keeps HTTP clients, HTTP servers, and background network tasks
4//! inside explicit resource budgets before they can allocate unbounded work.
5
6use std::{
7    error::Error,
8    fmt,
9    sync::{Arc, Mutex},
10};
11
12use crate::env::NetworkEnvOverrides;
13
14pub const DEFAULT_MAX_CONNECTIONS: usize = 1024;
15pub const DEFAULT_MAX_IN_FLIGHT_REQUESTS: usize = 256;
16pub const DEFAULT_MAX_QUEUE_DEPTH: usize = 512;
17
18/// Bounded resource policy for network admission decisions.
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct QosPolicy {
21    pub max_connections: usize,
22    pub max_in_flight_requests: usize,
23    pub max_queue_depth: usize,
24}
25
26impl QosPolicy {
27    /// Creates a policy and rejects zero-sized budgets.
28    pub fn new(
29        max_connections: usize,
30        max_in_flight_requests: usize,
31        max_queue_depth: usize,
32    ) -> Result<Self, QosPolicyError> {
33        ensure_positive("max_connections", max_connections)?;
34        ensure_positive("max_in_flight_requests", max_in_flight_requests)?;
35        ensure_positive("max_queue_depth", max_queue_depth)?;
36
37        Ok(Self {
38            max_connections,
39            max_in_flight_requests,
40            max_queue_depth,
41        })
42    }
43
44    /// Applies environment overrides to the default interactive QoS budgets.
45    pub fn from_overrides(overrides: &NetworkEnvOverrides) -> Result<Self, QosPolicyError> {
46        Self::new(
47            overrides
48                .qos_max_connections
49                .unwrap_or(DEFAULT_MAX_CONNECTIONS),
50            overrides
51                .qos_max_in_flight_requests
52                .unwrap_or(DEFAULT_MAX_IN_FLIGHT_REQUESTS),
53            overrides
54                .qos_max_queue_depth
55                .unwrap_or(DEFAULT_MAX_QUEUE_DEPTH),
56        )
57    }
58
59    /// Evaluates current resource usage before admitting network work.
60    pub fn evaluate(&self, snapshot: QosSnapshot) -> AdmissionDecision {
61        if snapshot.connections >= self.max_connections {
62            return AdmissionDecision::Reject(RejectReason::ConnectionBudgetExceeded);
63        }
64
65        if snapshot.in_flight_requests >= self.max_in_flight_requests {
66            return AdmissionDecision::Reject(RejectReason::RequestBudgetExceeded);
67        }
68
69        if snapshot.queued_requests >= self.max_queue_depth {
70            return AdmissionDecision::Reject(RejectReason::QueueBudgetExceeded);
71        }
72
73        AdmissionDecision::Admit
74    }
75}
76
77/// Point-in-time network resource usage for QoS decisions.
78#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
79pub struct QosSnapshot {
80    pub connections: usize,
81    pub in_flight_requests: usize,
82    pub queued_requests: usize,
83}
84
85/// Current QoS usage plus cumulative admission and overload counters.
86#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
87pub struct QosDiagnosticsSnapshot {
88    pub usage: QosSnapshot,
89    pub admitted_total: u64,
90    pub queued_total: u64,
91    pub rejected_total: u64,
92    pub timed_out_total: u64,
93    pub cancelled_total: u64,
94    pub dropped_total: u64,
95}
96
97/// Result of a QoS admission check.
98#[derive(Debug, Clone, Copy, PartialEq, Eq)]
99pub enum AdmissionDecision {
100    Admit,
101    Reject(RejectReason),
102}
103
104/// Reason attached to a rejected network operation.
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub enum RejectReason {
107    ConnectionBudgetExceeded,
108    RequestBudgetExceeded,
109    QueueBudgetExceeded,
110}
111
112impl RejectReason {
113    /// Stable low-cardinality reason label for diagnostics and metrics.
114    pub const fn as_str(self) -> &'static str {
115        match self {
116            Self::ConnectionBudgetExceeded => "connection_budget_exceeded",
117            Self::RequestBudgetExceeded => "request_budget_exceeded",
118            Self::QueueBudgetExceeded => "queue_budget_exceeded",
119        }
120    }
121}
122
123/// Runtime QoS counters used by inbound protocol adapters.
124#[derive(Debug, Clone, Default)]
125pub struct QosRuntime {
126    state: Arc<Mutex<QosState>>,
127}
128
129#[derive(Debug, Clone, Copy, Default)]
130struct QosState {
131    usage: QosSnapshot,
132    counters: QosCounters,
133}
134
135#[derive(Debug, Clone, Copy, Default)]
136struct QosCounters {
137    admitted_total: u64,
138    queued_total: u64,
139    rejected_total: u64,
140    timed_out_total: u64,
141    cancelled_total: u64,
142    dropped_total: u64,
143}
144
145impl QosRuntime {
146    /// Reserves space in the bounded request queue before active admission.
147    pub fn reserve_queue(&self, policy: &QosPolicy) -> Result<QosPermit, RejectReason> {
148        let mut state = self
149            .state
150            .lock()
151            .unwrap_or_else(|poisoned| poisoned.into_inner());
152        if state.usage.queued_requests >= policy.max_queue_depth {
153            state.record_rejection();
154            return Err(RejectReason::QueueBudgetExceeded);
155        }
156
157        state.usage.queued_requests += 1;
158        state.counters.queued_total = state.counters.queued_total.saturating_add(1);
159        Ok(QosPermit {
160            runtime: self.clone(),
161            kind: QosPermitKind::Queue,
162            released: false,
163        })
164    }
165
166    /// Applies queue and active request budgets atomically for immediate admission.
167    pub fn admit_queued_request(&self, policy: &QosPolicy) -> Result<QosPermit, RejectReason> {
168        let mut state = self
169            .state
170            .lock()
171            .unwrap_or_else(|poisoned| poisoned.into_inner());
172        if state.usage.queued_requests >= policy.max_queue_depth {
173            state.record_rejection();
174            return Err(RejectReason::QueueBudgetExceeded);
175        }
176
177        if state.usage.in_flight_requests >= policy.max_in_flight_requests {
178            state.record_rejection();
179            return Err(RejectReason::RequestBudgetExceeded);
180        }
181
182        state.usage.in_flight_requests += 1;
183        state.counters.queued_total = state.counters.queued_total.saturating_add(1);
184        state.counters.admitted_total = state.counters.admitted_total.saturating_add(1);
185        Ok(QosPermit {
186            runtime: self.clone(),
187            kind: QosPermitKind::Request,
188            released: false,
189        })
190    }
191
192    /// Attempts to admit one inbound request and returns a release-on-drop permit.
193    pub fn admit_request(&self, policy: &QosPolicy) -> Result<QosPermit, RejectReason> {
194        let mut state = self
195            .state
196            .lock()
197            .unwrap_or_else(|poisoned| poisoned.into_inner());
198        if state.usage.in_flight_requests >= policy.max_in_flight_requests {
199            state.record_rejection();
200            return Err(RejectReason::RequestBudgetExceeded);
201        }
202
203        state.usage.in_flight_requests += 1;
204        state.counters.admitted_total = state.counters.admitted_total.saturating_add(1);
205        Ok(QosPermit {
206            runtime: self.clone(),
207            kind: QosPermitKind::Request,
208            released: false,
209        })
210    }
211
212    /// Attempts to admit one open network connection.
213    pub fn admit_connection(&self, policy: &QosPolicy) -> Result<QosPermit, RejectReason> {
214        let mut state = self
215            .state
216            .lock()
217            .unwrap_or_else(|poisoned| poisoned.into_inner());
218        if state.usage.connections >= policy.max_connections {
219            state.record_rejection();
220            return Err(RejectReason::ConnectionBudgetExceeded);
221        }
222
223        state.usage.connections += 1;
224        state.counters.admitted_total = state.counters.admitted_total.saturating_add(1);
225        Ok(QosPermit {
226            runtime: self.clone(),
227            kind: QosPermitKind::Connection,
228            released: false,
229        })
230    }
231
232    /// Returns the current request budget snapshot for diagnostics and tests.
233    pub fn snapshot(&self) -> QosSnapshot {
234        self.state
235            .lock()
236            .unwrap_or_else(|poisoned| poisoned.into_inner())
237            .usage
238    }
239
240    /// Returns current usage and cumulative overload counters.
241    pub fn diagnostics_snapshot(&self) -> QosDiagnosticsSnapshot {
242        let state = self
243            .state
244            .lock()
245            .unwrap_or_else(|poisoned| poisoned.into_inner());
246        QosDiagnosticsSnapshot {
247            usage: state.usage,
248            admitted_total: state.counters.admitted_total,
249            queued_total: state.counters.queued_total,
250            rejected_total: state.counters.rejected_total,
251            timed_out_total: state.counters.timed_out_total,
252            cancelled_total: state.counters.cancelled_total,
253            dropped_total: state.counters.dropped_total,
254        }
255    }
256
257    /// Records an operation that exceeded its runtime timeout after admission.
258    pub fn record_timed_out(&self) {
259        let mut state = self
260            .state
261            .lock()
262            .unwrap_or_else(|poisoned| poisoned.into_inner());
263        state.counters.timed_out_total = state.counters.timed_out_total.saturating_add(1);
264    }
265
266    /// Records an admitted operation cancelled by the caller or peer.
267    pub fn record_cancelled(&self) {
268        let mut state = self
269            .state
270            .lock()
271            .unwrap_or_else(|poisoned| poisoned.into_inner());
272        state.counters.cancelled_total = state.counters.cancelled_total.saturating_add(1);
273    }
274
275    /// Records work dropped before application handling could start.
276    pub fn record_dropped(&self) {
277        let mut state = self
278            .state
279            .lock()
280            .unwrap_or_else(|poisoned| poisoned.into_inner());
281        state.counters.dropped_total = state.counters.dropped_total.saturating_add(1);
282    }
283
284    fn release(&self, kind: QosPermitKind) {
285        if matches!(kind, QosPermitKind::Noop) {
286            return;
287        }
288        let mut state = self
289            .state
290            .lock()
291            .unwrap_or_else(|poisoned| poisoned.into_inner());
292        match kind {
293            QosPermitKind::Connection => {
294                state.usage.connections = state.usage.connections.saturating_sub(1);
295            }
296            QosPermitKind::Request => {
297                state.usage.in_flight_requests = state.usage.in_flight_requests.saturating_sub(1);
298            }
299            QosPermitKind::Queue => {
300                state.usage.queued_requests = state.usage.queued_requests.saturating_sub(1);
301            }
302            QosPermitKind::Noop => {}
303        }
304    }
305}
306
307impl QosState {
308    fn record_rejection(&mut self) {
309        self.counters.rejected_total = self.counters.rejected_total.saturating_add(1);
310    }
311}
312
313#[derive(Debug, Clone, Copy)]
314enum QosPermitKind {
315    Connection,
316    Request,
317    Queue,
318    Noop,
319}
320
321/// Admission permit that releases its QoS budget on drop.
322#[derive(Debug)]
323pub struct QosPermit {
324    runtime: QosRuntime,
325    kind: QosPermitKind,
326    released: bool,
327}
328
329impl QosPermit {
330    /// Creates a permit for nested work already charged to an outer QoS boundary.
331    pub(crate) fn already_admitted(runtime: QosRuntime) -> Self {
332        Self {
333            runtime,
334            kind: QosPermitKind::Noop,
335            released: false,
336        }
337    }
338}
339
340impl Drop for QosPermit {
341    fn drop(&mut self) {
342        if !self.released {
343            self.runtime.release(self.kind);
344            self.released = true;
345        }
346    }
347}
348
349/// QoS policy validation error.
350#[derive(Debug, Clone, PartialEq, Eq)]
351pub struct QosPolicyError {
352    pub field: &'static str,
353}
354
355impl fmt::Display for QosPolicyError {
356    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
357        write!(formatter, "{} must be greater than zero", self.field)
358    }
359}
360
361impl Error for QosPolicyError {}
362
363fn ensure_positive(field: &'static str, value: usize) -> Result<(), QosPolicyError> {
364    if value == 0 {
365        return Err(QosPolicyError { field });
366    }
367
368    Ok(())
369}
370
371#[cfg(test)]
372#[path = "mod_tests.rs"]
373mod tests;
relay_knowledge/net/qos/mod.rs

relay_knowledge/net/qos/
mod.rs
