xeiaso.net
Proof of concept mechanical port of ircnet/ircd to Rust as part of a bit about C being insecure for network services
1//! Per-connection traffic counters (P11 slice 145).
2//!
3//! Each accepted connection owns a [`ConnStats`]: monotonic byte/message counters for
4//! data sent to and received from the socket, plus the connection's birth instant. The
5//! struct is shared between two tasks via `Arc`:
6//!
7//! - the connection's own serve loop (`serve_client` in `main.rs`) is the **writer** — it
8//! bumps the counters at every socket read/write; and
9//! - any *other* connection running `STATS l` is a **reader** — it enumerates the registry
10//! ([`crate::registry::Registry::link_info`]) and snapshots each link's counters into a
11//! `211 RPL_STATSLINKINFO` line.
12//!
13//! The counters are plain [`AtomicU64`]s with `Relaxed` ordering: the reader only needs an
14//! eventually-consistent snapshot for a stats dump, never a happens-before relationship
15//! with the wire, so the cheapest ordering is correct. The struct is lock-free, so the
16//! pure render path ([`crate::command::stats`]) and the fuzz tests drive it directly.
17
18use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
19
20/// The default SendQ ceiling (P11 slice 147) applied to a local client whose connection class
21/// does not set a `sendq` (the open-server / `sendq 0` case). 1 MiB — a generous client
22/// ceiling, above leveva's typical configured values. A SendQ ceiling is a safety mechanism
23/// (it bounds the unbounded mailbox under a non-draining peer), so unlike the slice-136
24/// per-class connection caps it is **always** on for a real client; only remote-user / test
25/// claims (which have no serve loop draining their mailbox) opt out with `sendq_max == 0`.
26pub const DEFAULT_SENDQ: u64 = 1_048_576;
27
28/// The default SendQ ceiling applied to a **server link** (P11 slice 148). 8 MiB — far above
29/// the client [`DEFAULT_SENDQ`] because a peer legitimately queues a whole network burst
30/// (every server + user + channel) on link-up, so a client-sized ceiling would kill a healthy
31/// burst (the `golden_s2s_mass_burst_drop` trap). It is the safety bound that keeps a
32/// non-draining peer's mailbox from growing without limit, not a per-connect tuning knob —
33/// resolving the matched `connect`/`class` block's `sendq` is a follow-on. Matches the shipped
34/// `class "servers" { sendq 8000000 }` in `dist/ircd.kdl`.
35pub const DEFAULT_SERVER_SENDQ: u64 = 8_000_000;
36
37/// The outcome of offering one frame to a connection's SendQ ([`ConnStats::sendq_admit`]).
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum SendqVerdict {
40 /// Under the limit — the frame's length was added to the queue depth; send it.
41 Send,
42 /// This frame pushes the depth over `max`. The add was backed out and the connection is
43 /// now latched dead — the caller queues **one** terminal eject instead, then drops the
44 /// rest. Returned to exactly one producer (the latch is a `swap`).
45 Overflow,
46 /// The connection is already latched dead (the terminal eject is in flight) — drop the
47 /// frame silently, without accounting, so a wedged socket can't grow the mailbox.
48 Dropped,
49}
50
51/// Live, lock-free traffic counters for one connection. Created at connection birth (in
52/// [`crate::session::Session::new`]); a clone of the `Arc` is handed to the registry at
53/// registration finalize so `STATS l` can read it from another task.
54#[derive(Debug)]
55pub struct ConnStats {
56 sent_messages: AtomicU64,
57 sent_bytes: AtomicU64,
58 recv_messages: AtomicU64,
59 recv_bytes: AtomicU64,
60 /// Bytes currently queued on the connection's mailbox: incremented by the registry when a
61 /// producer pushes a frame, decremented by the serve loop when it drains one (P11 slice
62 /// 147). The live send-queue depth `STATS l` reports.
63 sendq: AtomicU64,
64 /// High-water mark of [`Self::sendq`] (informational; not wire-rendered).
65 sendq_peak: AtomicU64,
66 /// Latched once the SendQ overflows its ceiling, so exactly one terminal eject is queued
67 /// and every later frame is dropped.
68 sendq_dead: AtomicBool,
69 /// Connection birth, unix **nanoseconds** ([`crate::clock::unixnano`]). Seeded once and
70 /// never moved; rendered as "seconds open" in `211`.
71 connected_at: u64,
72}
73
74/// A `Copy` point-in-time view of a [`ConnStats`], taken under no lock. The numbers may be
75/// momentarily stale relative to the live socket — fine for a stats dump.
76#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
77pub struct ConnStatsSnapshot {
78 pub sent_messages: u64,
79 pub sent_bytes: u64,
80 pub recv_messages: u64,
81 pub recv_bytes: u64,
82 /// Live send-queue depth in bytes (P11 slice 147) — what `STATS l`'s `<sendq>` reports.
83 pub sendq: u64,
84 pub connected_at: u64,
85}
86
87impl ConnStats {
88 /// A fresh counter set born at `connected_at` (unix-nanos). All four counters start at
89 /// zero.
90 pub fn new(connected_at: u64) -> Self {
91 Self {
92 sent_messages: AtomicU64::new(0),
93 sent_bytes: AtomicU64::new(0),
94 recv_messages: AtomicU64::new(0),
95 recv_bytes: AtomicU64::new(0),
96 sendq: AtomicU64::new(0),
97 sendq_peak: AtomicU64::new(0),
98 sendq_dead: AtomicBool::new(false),
99 connected_at,
100 }
101 }
102
103 /// Offer one `len`-byte frame to the send queue under ceiling `max` (P11 slice 147).
104 /// `max == 0` disables the kill (account-only — used for remote-user / test links that
105 /// have no serve loop draining their mailbox). See [`SendqVerdict`].
106 pub fn sendq_admit(&self, len: u64, max: u64) -> SendqVerdict {
107 if self.sendq_dead.load(Ordering::Relaxed) {
108 return SendqVerdict::Dropped;
109 }
110 let new = self.sendq.fetch_add(len, Ordering::Relaxed) + len;
111 if max != 0 && new > max {
112 // Over the ceiling: back the add out and latch dead. `swap` makes exactly one
113 // racing producer observe the transition (it emits the terminal eject); any other
114 // simultaneously-overflowing producer sees the latch already set and drops.
115 self.sendq.fetch_sub(len, Ordering::Relaxed);
116 if self.sendq_dead.swap(true, Ordering::Relaxed) {
117 return SendqVerdict::Dropped;
118 }
119 return SendqVerdict::Overflow;
120 }
121 self.sendq_peak.fetch_max(new, Ordering::Relaxed);
122 SendqVerdict::Send
123 }
124
125 /// Account `len` bytes onto the queue **unconditionally**, bypassing the dead latch — for
126 /// the terminal eject itself and for control-plane frames (KILL / collision SAVE) that are
127 /// always delivered (P11 slice 147).
128 pub fn sendq_force_add(&self, len: u64) {
129 let new = self.sendq.fetch_add(len, Ordering::Relaxed) + len;
130 self.sendq_peak.fetch_max(new, Ordering::Relaxed);
131 }
132
133 /// The serve loop drained one `len`-byte frame off the mailbox; drop the queue depth.
134 /// Saturating, so any accounting imbalance can never wrap the unsigned counter.
135 pub fn sendq_drain(&self, len: u64) {
136 let mut cur = self.sendq.load(Ordering::Relaxed);
137 loop {
138 let next = cur.saturating_sub(len);
139 match self
140 .sendq
141 .compare_exchange_weak(cur, next, Ordering::Relaxed, Ordering::Relaxed)
142 {
143 Ok(_) => return,
144 Err(actual) => cur = actual,
145 }
146 }
147 }
148
149 /// The current send-queue depth in bytes.
150 pub fn sendq_len(&self) -> u64 {
151 self.sendq.load(Ordering::Relaxed)
152 }
153
154 /// Whether this connection has been latched dead by a SendQ overflow.
155 pub fn sendq_is_dead(&self) -> bool {
156 self.sendq_dead.load(Ordering::Relaxed)
157 }
158
159 /// Account a chunk written to the socket: `bytes.len()` bytes and
160 /// [`count_messages`]`(bytes)` messages.
161 pub fn record_sent(&self, bytes: &[u8]) {
162 self.sent_bytes
163 .fetch_add(bytes.len() as u64, Ordering::Relaxed);
164 self.sent_messages
165 .fetch_add(count_messages(bytes), Ordering::Relaxed);
166 }
167
168 /// Account a chunk read from the socket: `bytes.len()` bytes and
169 /// [`count_messages`]`(bytes)` messages.
170 pub fn record_recv(&self, bytes: &[u8]) {
171 self.recv_bytes
172 .fetch_add(bytes.len() as u64, Ordering::Relaxed);
173 self.recv_messages
174 .fetch_add(count_messages(bytes), Ordering::Relaxed);
175 }
176
177 /// Take a consistent-enough snapshot for a stats line.
178 pub fn snapshot(&self) -> ConnStatsSnapshot {
179 ConnStatsSnapshot {
180 sent_messages: self.sent_messages.load(Ordering::Relaxed),
181 sent_bytes: self.sent_bytes.load(Ordering::Relaxed),
182 recv_messages: self.recv_messages.load(Ordering::Relaxed),
183 recv_bytes: self.recv_bytes.load(Ordering::Relaxed),
184 sendq: self.sendq.load(Ordering::Relaxed),
185 connected_at: self.connected_at,
186 }
187 }
188}
189
190/// Count IRC messages in a wire chunk: the number of `\n` line terminators. A partial line
191/// with no terminator yet contributes nothing (it is counted when its `\n` arrives in a
192/// later chunk), matching how the framer completes a line. Well-formed leveva output is
193/// CRLF-terminated, so this equals the line count.
194pub fn count_messages(bytes: &[u8]) -> u64 {
195 bytes.iter().filter(|&&b| b == b'\n').count() as u64
196}
197
198#[cfg(test)]
199mod tests {
200 use super::*;
201
202 #[test]
203 fn fresh_stats_are_all_zero() {
204 let s = ConnStats::new(42);
205 let snap = s.snapshot();
206 assert_eq!(snap.sent_messages, 0);
207 assert_eq!(snap.sent_bytes, 0);
208 assert_eq!(snap.recv_messages, 0);
209 assert_eq!(snap.recv_bytes, 0);
210 assert_eq!(snap.connected_at, 42);
211 }
212
213 #[test]
214 fn record_sent_accumulates_bytes_and_messages() {
215 let s = ConnStats::new(0);
216 s.record_sent(b"PING :x\r\n");
217 s.record_sent(b"PONG :y\r\nNOTICE a :hi\r\n"); // two lines
218 let snap = s.snapshot();
219 assert_eq!(snap.sent_bytes, 9 + 23);
220 assert_eq!(snap.sent_messages, 1 + 2);
221 // recv untouched — the two directions are independent.
222 assert_eq!(snap.recv_bytes, 0);
223 assert_eq!(snap.recv_messages, 0);
224 }
225
226 #[test]
227 fn record_recv_accumulates_independently() {
228 let s = ConnStats::new(0);
229 s.record_recv(b"NICK alice\r\n");
230 let snap = s.snapshot();
231 assert_eq!(snap.recv_bytes, 12);
232 assert_eq!(snap.recv_messages, 1);
233 assert_eq!(snap.sent_bytes, 0);
234 }
235
236 #[test]
237 fn sendq_admit_under_limit_accounts_and_sends() {
238 let s = ConnStats::new(0);
239 assert_eq!(s.sendq_admit(100, 1000), SendqVerdict::Send);
240 assert_eq!(s.sendq_admit(100, 1000), SendqVerdict::Send);
241 assert_eq!(s.sendq_len(), 200);
242 assert!(!s.sendq_is_dead());
243 assert_eq!(s.snapshot().sendq, 200);
244 }
245
246 #[test]
247 fn sendq_admit_over_limit_overflows_latches_and_backs_out_the_add() {
248 let s = ConnStats::new(0);
249 assert_eq!(s.sendq_admit(800, 1000), SendqVerdict::Send);
250 // 800 + 300 = 1100 > 1000 → Overflow; the 300 is backed out (depth stays 800).
251 assert_eq!(s.sendq_admit(300, 1000), SendqVerdict::Overflow);
252 assert_eq!(s.sendq_len(), 800);
253 assert!(s.sendq_is_dead());
254 // Once dead, every later frame is Dropped without accounting.
255 assert_eq!(s.sendq_admit(1, 1000), SendqVerdict::Dropped);
256 assert_eq!(s.sendq_len(), 800);
257 }
258
259 #[test]
260 fn sendq_admit_exactly_at_limit_is_send() {
261 let s = ConnStats::new(0);
262 // new == max is allowed (only `> max` overflows).
263 assert_eq!(s.sendq_admit(1000, 1000), SendqVerdict::Send);
264 assert!(!s.sendq_is_dead());
265 assert_eq!(s.sendq_admit(1, 1000), SendqVerdict::Overflow);
266 }
267
268 #[test]
269 fn sendq_max_zero_never_overflows() {
270 let s = ConnStats::new(0);
271 for _ in 0..1000 {
272 assert_eq!(s.sendq_admit(10_000, 0), SendqVerdict::Send);
273 }
274 assert!(!s.sendq_is_dead());
275 assert_eq!(s.sendq_len(), 10_000_000);
276 }
277
278 #[test]
279 fn sendq_drain_decrements_and_saturates() {
280 let s = ConnStats::new(0);
281 s.sendq_admit(500, 0);
282 s.sendq_drain(200);
283 assert_eq!(s.sendq_len(), 300);
284 // Inverse: draining more than is queued floors at zero (never wraps).
285 s.sendq_drain(10_000);
286 assert_eq!(s.sendq_len(), 0);
287 }
288
289 #[test]
290 fn sendq_force_add_bypasses_the_dead_latch() {
291 let s = ConnStats::new(0);
292 s.sendq_admit(1000, 1000);
293 s.sendq_admit(1, 1000); // Overflow → latched dead
294 assert!(s.sendq_is_dead());
295 // A control-plane frame still accounts even past death.
296 s.sendq_force_add(50);
297 assert_eq!(s.sendq_len(), 1050);
298 }
299
300 #[test]
301 fn sendq_enqueue_drain_round_trips_to_zero() {
302 // Inverse invariant: a balanced enqueue/drain sequence returns depth to exactly 0.
303 let s = ConnStats::new(0);
304 for n in [10u64, 20, 30, 40] {
305 s.sendq_admit(n, 0);
306 }
307 for n in [10u64, 20, 30, 40] {
308 s.sendq_drain(n);
309 }
310 assert_eq!(s.sendq_len(), 0);
311 }
312
313 #[test]
314 fn count_messages_is_the_newline_count() {
315 assert_eq!(count_messages(b""), 0);
316 assert_eq!(count_messages(b"no terminator"), 0); // partial line, not yet a message
317 assert_eq!(count_messages(b"one\r\n"), 1);
318 assert_eq!(count_messages(b"a\r\nb\r\nc\r\n"), 3);
319 // a trailing partial after a complete line counts only the complete one
320 assert_eq!(count_messages(b"done\r\npartial"), 1);
321 }
322}