krabka_client_streams/dsl/
suppress.rs

1//! `Suppressed` and `BufferConfig`: the suppress configuration surface.
2//!
3//! Slice A implements `until_window_closes(unbounded())`, which gives final
4//! results for windowed tables. Slice B adds `with_max_records`, a bounded
5//! buffer with `shutDownWhenFull`. Slice C adds `until_time_limit`, the
6//! `emit_early_when_full` overflow toggle, the eager `max_records(n)`
7//! constructor, `record_cap()`, and `is_emit_early()`. Slice D adds the logging
8//! toggle.
9
10use krabka_units::prelude::*;
11
12/// The bound on the suppress buffer and the behaviour when it is full.
13///
14/// A [`ByteSize`] stores `f64`, so this type is `PartialEq` but not `Eq`. Nothing
15/// keys a map or a set on a buffer config, so the weaker bound costs nothing.
16#[derive(Clone, Copy, Debug, PartialEq)]
17pub struct BufferConfig {
18    max_records: Option<usize>,
19    /// Cap on the total serialized size of the buffer, the sum of `key_bytes`
20    /// and `value_bytes`. The processor enforces it against the registered
21    /// store's `byte_size()`. This is the JVM `BufferConfig.maxBytes`.
22    max_bytes: Option<ByteSize>,
23    /// `false` is shutDownWhenFull, which is strict and panics. `true` is
24    /// emitEarlyWhenFull, which is eager.
25    emit_early: bool,
26}
27
28impl BufferConfig {
29    /// Unbounded and strict, that is, shutDownWhenFull.
30    #[must_use]
31    pub fn unbounded() -> Self {
32        Self {
33            max_records: None,
34            max_bytes: None,
35            emit_early: false,
36        }
37    }
38
39    /// Cap at `n` records, EAGER, that is, emit-early-when-full.
40    ///
41    /// This matches the JVM static `BufferConfig.maxRecords(n)`, the
42    /// rate-limiter default overflow.
43    #[must_use]
44    /// # Panics
45    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
46    pub fn max_records(n: usize) -> Self {
47        assert!(n >= 1, "max_records must be >= 1");
48        Self {
49            max_records: Some(n),
50            max_bytes: None,
51            emit_early: true,
52        }
53    }
54
55    /// Cap at `n`, EAGER, that is, emit-early-when-full.
56    ///
57    /// This matches the JVM static `BufferConfig.maxBytes(n)`. The processor
58    /// measures the cap against the serialized `key_bytes` and `value_bytes`
59    /// summed across the buffered entries.
60    #[must_use]
61    /// # Panics
62    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
63    pub fn max_bytes(n: ByteSize) -> Self {
64        assert!(n >= bytes(1), "max_bytes must be >= 1");
65        Self {
66            max_records: None,
67            max_bytes: Some(n),
68            emit_early: true,
69        }
70    }
71
72    /// Cap at `n` records and keep the current overflow mode.
73    ///
74    /// The mode is strict on the `unbounded()` path. This matches the JVM
75    /// `unbounded().withMaxRecords(n)`.
76    #[must_use]
77    /// # Panics
78    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
79    pub fn with_max_records(self, n: usize) -> Self {
80        assert!(n >= 1, "max_records must be >= 1");
81        Self {
82            max_records: Some(n),
83            ..self
84        }
85    }
86
87    /// Cap at `n` and keep the current overflow mode. This matches the JVM
88    /// `unbounded().withMaxBytes(n)`.
89    #[must_use]
90    /// # Panics
91    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
92    pub fn with_max_bytes(self, n: ByteSize) -> Self {
93        assert!(n >= bytes(1), "max_bytes must be >= 1");
94        Self {
95            max_bytes: Some(n),
96            ..self
97        }
98    }
99
100    /// Evict and emit the oldest buffered record when the buffer is full. This
101    /// is the eager mode.
102    #[must_use]
103    pub fn emit_early_when_full(self) -> Self {
104        Self {
105            emit_early: true,
106            ..self
107        }
108    }
109
110    /// Shut the task down when the buffer is full. This is the strict mode.
111    #[must_use]
112    pub fn shut_down_when_full(self) -> Self {
113        Self {
114            emit_early: false,
115            ..self
116        }
117    }
118
119    pub(crate) fn record_cap(&self) -> Option<usize> {
120        self.max_records
121    }
122    pub(crate) fn byte_cap(&self) -> Option<ByteSize> {
123        self.max_bytes
124    }
125    pub(crate) fn is_emit_early(&self) -> bool {
126        self.emit_early
127    }
128}
129
130/// A suppression configuration, parameterized by the table key `K`.
131///
132/// It carries a `fn(&K, i64) -> i64` that maps a record key and a timestamp to a
133/// buffer time. Window-close reads `window.end`, and time-limit reads the record
134/// timestamp. Fn pointers are `Copy`, so `Suppressed<K>` is `Copy` and needs no
135/// `K: Copy` bound.
136#[derive(Debug)]
137pub struct Suppressed<K> {
138    pub(crate) buffer: BufferConfig,
139    pub(crate) buffer_time: fn(&K, i64) -> i64,
140    pub(crate) wait: WaitKind,
141    /// Whether the client emits the suppress buffer's changelog topic. Default:
142    /// `true`. `false` matches the JVM `withLoggingDisabled()` and keeps the
143    /// buffer in memory only. There is then no changelog topic in the wire
144    /// topology and no restore.
145    pub(crate) logging: bool,
146}
147
148// All fields are Copy independently of K (fn pointer + plain enums/bool).
149impl<K> Clone for Suppressed<K> {
150    fn clone(&self) -> Self {
151        *self
152    }
153}
154
155impl<K> Copy for Suppressed<K> {}
156
157impl<K> Suppressed<K> {
158    /// Disable the suppress buffer's changelog, as the JVM
159    /// `withLoggingDisabled()` does. The buffer then stays in memory only, with
160    /// no changelog topic and no fault-tolerance.
161    #[must_use]
162    pub fn with_logging_disabled(self) -> Self {
163        Self {
164            logging: false,
165            ..self
166        }
167    }
168
169    /// Re-enable the suppress buffer's changelog. This is the default.
170    #[must_use]
171    pub fn with_logging_enabled(self) -> Self {
172        Self {
173            logging: true,
174            ..self
175        }
176    }
177}
178
179/// How long to wait before emitting a buffered record.
180#[derive(Clone, Copy, Debug)]
181pub(crate) enum WaitKind {
182    /// Window-close. The wait is the upstream window's grace, taken from the
183    /// `KTable` handle.
184    UpstreamGrace,
185    /// Time-limit. The wait is the configured duration.
186    Fixed(Time),
187}
188
189impl<KInner> Suppressed<crate::dsl::windows::Windowed<KInner>> {
190    /// Emit each window's final result once the window closes, that is, once
191    /// `stream_time >= window.end + grace`. This needs a windowed `KTable` and a
192    /// STRICT buffer, that is, shutDownWhenFull.
193    #[must_use]
194    /// # Panics
195    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
196    pub fn until_window_closes(buffer: BufferConfig) -> Self {
197        assert!(
198            !buffer.is_emit_early(),
199            "untilWindowCloses requires a strict (shutDownWhenFull) buffer config"
200        );
201        Self {
202            buffer,
203            buffer_time: |k, _ts| k.window.end,
204            wait: WaitKind::UpstreamGrace,
205            logging: true,
206        }
207    }
208}
209
210impl<K> Suppressed<K> {
211    /// Rate-limiter: emit at most one update per key per `wait` in stream time.
212    /// A newer record for a key replaces the buffered one and resets the timer.
213    #[must_use]
214    /// # Panics
215    /// Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.
216    pub fn until_time_limit(wait: Time, buffer: BufferConfig) -> Self {
217        assert!(wait >= Time::ZERO, "time limit must be >= 0");
218        Self {
219            buffer,
220            buffer_time: |_k, ts| ts,
221            wait: WaitKind::Fixed(wait),
222            logging: true,
223        }
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use assert2::check;
230
231    use super::*;
232
233    #[test]
234    fn buffer_config_caps_and_overflow() {
235        check!(BufferConfig::unbounded().record_cap() == None);
236        check!(!BufferConfig::unbounded().is_emit_early()); // strict
237        let strict = BufferConfig::unbounded().with_max_records(3);
238        check!(strict.record_cap() == Some(3));
239        check!(!strict.is_emit_early());
240        let eager = BufferConfig::max_records(5); // eager
241        check!(eager.record_cap() == Some(5));
242        check!(eager.is_emit_early());
243        check!(!eager.shut_down_when_full().is_emit_early());
244        check!(
245            BufferConfig::unbounded()
246                .emit_early_when_full()
247                .is_emit_early()
248        );
249    }
250
251    #[test]
252    fn buffer_config_byte_caps() {
253        check!(BufferConfig::unbounded().byte_cap() == None);
254        let eager = BufferConfig::max_bytes(kibibytes(1)); // eager static
255        check!(eager.byte_cap() == Some(kibibytes(1)));
256        check!(eager.record_cap() == None);
257        check!(eager.is_emit_early());
258        // strict path keeps shutDownWhenFull
259        let strict = BufferConfig::unbounded().with_max_bytes(bytes(512));
260        check!(strict.byte_cap() == Some(bytes(512)));
261        check!(!strict.is_emit_early());
262        // records + bytes coexist
263        let both = BufferConfig::unbounded()
264            .with_max_records(3)
265            .with_max_bytes(bytes(99));
266        check!(both.record_cap() == Some(3));
267        check!(both.byte_cap() == Some(bytes(99)));
268    }
269
270    #[test]
271    fn logging_toggles() {
272        use crate::dsl::windows::{Window, Windowed};
273        let on = Suppressed::until_window_closes(BufferConfig::unbounded());
274        check!(on.logging); // default on
275        let off = on.with_logging_disabled();
276        check!(!off.logging);
277        check!(off.with_logging_enabled().logging);
278        // window-close buffer_time still reads window.end after the toggle
279        let wk = Windowed {
280            key: "k".to_string(),
281            window: Window { start: 0, end: 7 },
282        };
283        check!((off.buffer_time)(&wk, 1) == 7);
284    }
285
286    #[test]
287    fn suppressed_constructors() {
288        use crate::dsl::windows::{Window, Windowed};
289        let wc = Suppressed::until_window_closes(BufferConfig::unbounded());
290        let wk = Windowed {
291            key: "k".to_string(),
292            window: Window { start: 0, end: 99 },
293        };
294        check!((wc.buffer_time)(&wk, 5) == 99); // window.end
295        let tl = Suppressed::<String>::until_time_limit(millis(50), BufferConfig::max_records(2));
296        check!((tl.buffer_time)(&"k".to_string(), 5) == 5); // record ts
297    }
298}
krabka_client_streams/dsl/suppress.rs

krabka_client_streams/dsl/
suppress.rs
