krabka_units/
convert.rs

1//! Conversions between quantities and the raw numbers at the edges of the system.
2//!
3//! Quantities live in the hand-written domain layer. At its edges the value must
4//! be a plain integer, float, or [`core::time::Duration`]. Those edges are the
5//! generated Kafka codec, an on-disk record, a `tokio` timer, and a Prometheus
6//! gauge. These extension traits are that seam, so this module states the
7//! rounding and saturation rules once instead of at every boundary.
8//!
9//! Conversions *into* a quantity are exact for every magnitude Kafka can express.
10//! Conversions *out* round to nearest and saturate at the target type's bounds
11//! instead of wrapping, and they map `NaN` to zero.
12
13use core::{marker::PhantomData, time::Duration};
14
15use num_traits::cast::{NumCast, ToPrimitive};
16
17use crate::{ByteRate, ByteSize, Frequency, Ratio, Time};
18
19/// Rounds to nearest and saturates at `i64`'s bounds; `NaN` becomes `0`.
20fn round_i64(value: f64) -> i64 {
21    if value.is_nan() {
22        return 0;
23    }
24    let rounded = value.round();
25    rounded
26        .to_i64()
27        .unwrap_or(if rounded < 0.0 { i64::MIN } else { i64::MAX })
28}
29
30/// Rounds to nearest and saturates at `i32`'s bounds; `NaN` becomes `0`.
31fn round_i32(value: f64) -> i32 {
32    if value.is_nan() {
33        return 0;
34    }
35    let rounded = value.round();
36    rounded
37        .to_i32()
38        .unwrap_or(if rounded < 0.0 { i32::MIN } else { i32::MAX })
39}
40
41/// Rounds to nearest and saturates into `0..=u64::MAX`; negatives and `NaN`
42/// become `0`.
43fn round_u64(value: f64) -> u64 {
44    if value.is_nan() {
45        return 0;
46    }
47    let rounded = value.round();
48    rounded
49        .to_u64()
50        .unwrap_or(if rounded < 0.0 { 0 } else { u64::MAX })
51}
52
53/// Rounds to nearest and saturates into `0..=usize::MAX`; negatives and `NaN`
54/// become `0`.
55fn round_usize(value: f64) -> usize {
56    if value.is_nan() {
57        return 0;
58    }
59    let rounded = value.round();
60    rounded
61        .to_usize()
62        .unwrap_or(if rounded < 0.0 { 0 } else { usize::MAX })
63}
64
65/// Widens an integer into `f64`. Exact for every magnitude below 2^53.
66fn widen<T: ToPrimitive>(value: T) -> f64 {
67    NumCast::from(value).unwrap_or(f64::NAN)
68}
69
70/// Byte counts: construction from raw integers, and extraction back to them.
71pub trait ByteSizeExt: Sized + Copy {
72    /// No bytes.
73    const ZERO: Self;
74
75    /// A raw unsigned byte count.
76    fn from_bytes(bytes: u64) -> Self;
77
78    /// A raw signed byte count. This keeps the sign, so it also carries a
79    /// *delta*, for example a quota balance that has gone into deficit. This
80    /// function does not interpret Kafka's `-1` "unlimited" sentinel. Use
81    /// [`wire::opt_size_from_bytes_i64`] for that.
82    fn from_bytes_i64(bytes: i64) -> Self;
83
84    /// A raw byte count measured as a float (a sampled or averaged size).
85    fn from_bytes_f64(bytes: f64) -> Self;
86
87    /// The count as an unsigned byte total.
88    fn bytes_u64(self) -> u64;
89
90    /// The count as a `usize`, for buffer capacities and slice lengths.
91    fn bytes_usize(self) -> usize;
92
93    /// The count as Kafka's `int32` byte fields (`max_bytes`, `min_bytes`).
94    fn bytes_i32(self) -> i32;
95
96    /// The count as Kafka's `int64` byte fields (`retention.bytes`).
97    fn bytes_i64(self) -> i64;
98
99    /// The count as a float, for metrics gauges.
100    fn bytes_f64(self) -> f64;
101}
102
103impl ByteSizeExt for ByteSize {
104    const ZERO: Self = Self {
105        dimension: PhantomData,
106        units: PhantomData,
107        value: 0.0,
108    };
109
110    fn from_bytes(bytes: u64) -> Self {
111        Self::from_bytes_f64(widen(bytes))
112    }
113
114    fn from_bytes_i64(bytes: i64) -> Self {
115        Self::from_bytes_f64(widen(bytes))
116    }
117
118    fn from_bytes_f64(bytes: f64) -> Self {
119        Self {
120            dimension: PhantomData,
121            units: PhantomData,
122            value: bytes,
123        }
124    }
125
126    fn bytes_u64(self) -> u64 {
127        round_u64(self.value)
128    }
129
130    fn bytes_usize(self) -> usize {
131        round_usize(self.value)
132    }
133
134    fn bytes_i32(self) -> i32 {
135        round_i32(self.value)
136    }
137
138    fn bytes_i64(self) -> i64 {
139        round_i64(self.value)
140    }
141
142    fn bytes_f64(self) -> f64 {
143        self.value
144    }
145}
146
147/// Time extents: construction from raw integers and [`Duration`], and extraction
148/// back to them.
149pub trait TimeExt: Sized + Copy {
150    /// No elapsed time.
151    const ZERO: Self;
152
153    /// A raw millisecond count. This is Kafka's unit for every configured
154    /// timeout, interval, and retention window.
155    fn from_millis(millis: i64) -> Self;
156
157    /// A raw microsecond count.
158    fn from_micros(micros: i64) -> Self;
159
160    /// A raw nanosecond count, for measured latencies.
161    fn from_nanos(nanos: i64) -> Self;
162
163    /// A raw second count.
164    fn from_secs(secs: i64) -> Self;
165
166    /// A fractional second count.
167    fn from_secs_f64(secs: f64) -> Self;
168
169    /// A [`Duration`], which is the unit `tokio`, `std`, and `qubit-clock` use.
170    fn from_std(duration: Duration) -> Self;
171
172    /// The extent as Kafka's `int32` millisecond fields (`session_timeout_ms`).
173    fn millis_i32(self) -> i32;
174
175    /// The extent as Kafka's `int64` millisecond fields (`retention_ms`).
176    fn millis_i64(self) -> i64;
177
178    /// The extent in whole milliseconds, truncated toward zero.
179    ///
180    /// [`Self::millis_i64`] rounds to nearest, which is correct for a value that
181    /// a caller reads back as a duration. Use this function only to match an
182    /// external format that truncates. Tempo's `durationMs` is its nanosecond
183    /// duration integer-divided by a million, so rounding would report one
184    /// millisecond more than Tempo does for the same span.
185    fn millis_i64_trunc(self) -> i64;
186
187    /// The extent in microseconds.
188    fn micros_i64(self) -> i64;
189
190    /// The extent in nanoseconds.
191    fn nanos_i64(self) -> i64;
192
193    /// The extent in whole seconds, rounded to nearest.
194    fn secs_i64(self) -> i64;
195
196    /// The extent in fractional seconds, for metrics histograms.
197    fn secs_f64(self) -> f64;
198
199    /// The extent as a [`Duration`], for `tokio::time::sleep` and friends.
200    ///
201    /// A negative or `NaN` extent becomes [`Duration::ZERO`], because `Duration`
202    /// cannot represent it and every caller means "do not sleep". A deadline that
203    /// has already passed gives such an extent.
204    fn to_std(self) -> Duration;
205}
206
207impl TimeExt for Time {
208    const ZERO: Self = Self {
209        dimension: PhantomData,
210        units: PhantomData,
211        value: 0.0,
212    };
213
214    fn from_millis(millis: i64) -> Self {
215        Self::from_secs_f64(widen(millis) / 1e3)
216    }
217
218    fn from_micros(micros: i64) -> Self {
219        Self::from_secs_f64(widen(micros) / 1e6)
220    }
221
222    fn from_nanos(nanos: i64) -> Self {
223        Self::from_secs_f64(widen(nanos) / 1e9)
224    }
225
226    fn from_secs(secs: i64) -> Self {
227        Self::from_secs_f64(widen(secs))
228    }
229
230    fn from_secs_f64(secs: f64) -> Self {
231        Self {
232            dimension: PhantomData,
233            units: PhantomData,
234            value: secs,
235        }
236    }
237
238    fn from_std(duration: Duration) -> Self {
239        Self::from_secs_f64(duration.as_secs_f64())
240    }
241
242    fn millis_i32(self) -> i32 {
243        round_i32(self.value * 1e3)
244    }
245
246    fn millis_i64(self) -> i64 {
247        round_i64(self.value * 1e3)
248    }
249
250    fn millis_i64_trunc(self) -> i64 {
251        // Divide the exact nanosecond count rather than truncating a scaled
252        // float: `0.042 * 1e3` is not exactly 42, so a float `trunc` here would
253        // report 41 milliseconds for a 42-millisecond extent.
254        self.nanos_i64() / 1_000_000
255    }
256
257    fn micros_i64(self) -> i64 {
258        round_i64(self.value * 1e6)
259    }
260
261    fn nanos_i64(self) -> i64 {
262        round_i64(self.value * 1e9)
263    }
264
265    fn secs_i64(self) -> i64 {
266        round_i64(self.value)
267    }
268
269    fn secs_f64(self) -> f64 {
270        self.value
271    }
272
273    fn to_std(self) -> Duration {
274        Duration::try_from_secs_f64(self.value).unwrap_or(Duration::ZERO)
275    }
276}
277
278/// Byte throughputs: construction from raw integers, and extraction back to them.
279pub trait ByteRateExt: Sized + Copy {
280    /// No throughput.
281    const ZERO: Self;
282
283    /// A raw bytes-per-second rate, as Kafka's quota configs express it.
284    fn from_bytes_per_sec(rate: i64) -> Self;
285
286    /// A fractional bytes-per-second rate, as a measurement produces it.
287    fn from_bytes_per_sec_f64(rate: f64) -> Self;
288
289    /// The rate in bytes per second.
290    fn bytes_per_sec_f64(self) -> f64;
291
292    /// The rate as Kafka's `int64` quota fields.
293    fn bytes_per_sec_i64(self) -> i64;
294
295    /// How long transferring `size` takes at this rate.
296    ///
297    /// A zero or negative rate gives [`TimeExt::ZERO`] and not an infinity,
298    /// because every caller reads an unset quota as "unlimited".
299    fn time_to_transfer(self, size: ByteSize) -> Time;
300}
301
302impl ByteRateExt for ByteRate {
303    const ZERO: Self = Self {
304        dimension: PhantomData,
305        units: PhantomData,
306        value: 0.0,
307    };
308
309    fn from_bytes_per_sec(rate: i64) -> Self {
310        Self::from_bytes_per_sec_f64(widen(rate))
311    }
312
313    fn from_bytes_per_sec_f64(rate: f64) -> Self {
314        Self {
315            dimension: PhantomData,
316            units: PhantomData,
317            value: rate,
318        }
319    }
320
321    fn bytes_per_sec_f64(self) -> f64 {
322        self.value
323    }
324
325    fn bytes_per_sec_i64(self) -> i64 {
326        round_i64(self.value)
327    }
328
329    fn time_to_transfer(self, size: ByteSize) -> Time {
330        if self.value <= 0.0 {
331            return <Time as TimeExt>::ZERO;
332        }
333        size / self
334    }
335}
336
337/// Event rates: construction from raw numbers, and extraction back to them.
338pub trait FrequencyExt: Sized + Copy {
339    /// No events.
340    const ZERO: Self;
341
342    /// A raw events-per-second rate.
343    fn from_per_sec(rate: f64) -> Self;
344
345    /// A whole-events-per-second rate, as a counter-based limiter configures one.
346    fn from_per_sec_u64(rate: u64) -> Self;
347
348    /// The rate in events per second.
349    fn per_sec_f64(self) -> f64;
350
351    /// The rate as whole events per second, for a limiter that counts in
352    /// integers. Rounds to nearest; negatives and `NaN` become zero.
353    fn per_sec_u64(self) -> u64;
354
355    /// The interval between events at this rate.
356    ///
357    /// A zero or negative rate gives [`TimeExt::ZERO`] and not an infinity.
358    fn period(self) -> Time;
359}
360
361impl FrequencyExt for Frequency {
362    const ZERO: Self = Self {
363        dimension: PhantomData,
364        units: PhantomData,
365        value: 0.0,
366    };
367
368    fn from_per_sec(rate: f64) -> Self {
369        Self {
370            dimension: PhantomData,
371            units: PhantomData,
372            value: rate,
373        }
374    }
375
376    fn from_per_sec_u64(rate: u64) -> Self {
377        Self::from_per_sec(widen(rate))
378    }
379
380    fn per_sec_f64(self) -> f64 {
381        self.value
382    }
383
384    fn per_sec_u64(self) -> u64 {
385        round_u64(self.value)
386    }
387
388    fn period(self) -> Time {
389        if self.value <= 0.0 {
390            return <Time as TimeExt>::ZERO;
391        }
392        1.0 / self
393    }
394}
395
396/// Dimensionless fractions.
397pub trait RatioExt: Sized + Copy {
398    /// Nothing.
399    const ZERO: Self;
400
401    /// The whole (`1.0`, or 100%).
402    const ONE: Self;
403
404    /// The fraction as a plain number, where `1.0` is the whole.
405    fn as_f64(self) -> f64;
406
407    /// The fraction as a percentage.
408    fn percent_f64(self) -> f64;
409}
410
411impl RatioExt for Ratio {
412    const ZERO: Self = Self {
413        dimension: PhantomData,
414        units: PhantomData,
415        value: 0.0,
416    };
417
418    const ONE: Self = Self {
419        dimension: PhantomData,
420        units: PhantomData,
421        value: 1.0,
422    };
423
424    fn as_f64(self) -> f64 {
425        self.value
426    }
427
428    fn percent_f64(self) -> f64 {
429        self.value * 100.0
430    }
431}
432
433/// Converting a [`Duration`] at an API boundary that hands one out.
434pub trait StdDurationExt {
435    /// This duration as a [`Time`] quantity.
436    fn as_time(&self) -> Time;
437}
438
439impl StdDurationExt for Duration {
440    fn as_time(&self) -> Time {
441        Time::from_std(*self)
442    }
443}
444
445/// Kafka's `-1` sentinel, which means "unlimited", "none", or "no bound"
446/// depending on the field, mapped to and from [`Option`].
447///
448/// A `None` means the field is unset, and the caller decides what unset implies.
449/// Do not use these functions for fields where `-1` means something else. A
450/// `timeout_ms` of `-1` means "block indefinitely", which is a value and not an
451/// absence, and the calling code's own enum models it better.
452pub mod wire {
453    use super::{ByteRate, ByteRateExt, ByteSize, ByteSizeExt, Time, TimeExt};
454
455    /// `-1` becomes `None`; any other value is a byte count.
456    #[must_use]
457    pub fn opt_size_from_bytes_i32(raw: i32) -> Option<ByteSize> {
458        (raw >= 0).then(|| ByteSize::from_bytes_i64(i64::from(raw)))
459    }
460
461    /// `None` becomes `-1`.
462    #[must_use]
463    pub fn opt_size_to_bytes_i32(size: Option<ByteSize>) -> i32 {
464        size.map_or(-1, ByteSizeExt::bytes_i32)
465    }
466
467    /// `-1` becomes `None`; any other value is a byte count.
468    #[must_use]
469    pub fn opt_size_from_bytes_i64(raw: i64) -> Option<ByteSize> {
470        (raw >= 0).then(|| ByteSize::from_bytes_i64(raw))
471    }
472
473    /// `None` becomes `-1`.
474    #[must_use]
475    pub fn opt_size_to_bytes_i64(size: Option<ByteSize>) -> i64 {
476        size.map_or(-1, ByteSizeExt::bytes_i64)
477    }
478
479    /// `-1` becomes `None`; any other value is a millisecond extent.
480    #[must_use]
481    pub fn opt_time_from_millis_i64(raw: i64) -> Option<Time> {
482        (raw >= 0).then(|| Time::from_millis(raw))
483    }
484
485    /// `None` becomes `-1`.
486    #[must_use]
487    pub fn opt_time_to_millis_i64(time: Option<Time>) -> i64 {
488        time.map_or(-1, TimeExt::millis_i64)
489    }
490
491    /// `-1` becomes `None`; any other value is a bytes-per-second quota.
492    #[must_use]
493    pub fn opt_rate_from_bytes_per_sec_i64(raw: i64) -> Option<ByteRate> {
494        (raw >= 0).then(|| ByteRate::from_bytes_per_sec(raw))
495    }
496
497    /// `None` becomes `-1`.
498    #[must_use]
499    pub fn opt_rate_to_bytes_per_sec_i64(rate: Option<ByteRate>) -> i64 {
500        rate.map_or(-1, ByteRateExt::bytes_per_sec_i64)
501    }
502}
krabka_units/convert.rs

krabka_units/
convert.rs
