spate_core/sink/config.rs
1//! Sink worker-pool tuning knobs.
2//!
3//! These are framework-level structs, and they are also the wire format: the
4//! `batch`, `inflight`, `retry` and `breaker` sub-sections of every sink's
5//! YAML deserialize straight into them. The keys and defaults are meant to be
6//! identical across connectors, so they are literally one type rather than a
7//! per-connector mirror that has to be kept in step by hand.
8//!
9//! Every struct here is `#[non_exhaustive]`. Construct one with `default()`
10//! (or [`SinkPoolConfig::new`]) and assign the fields you are tuning; a knob
11//! added later arrives as a new default and existing code keeps compiling.
12
13use bytesize::ByteSize;
14use serde::{Deserialize, Deserializer};
15use std::time::Duration;
16
17/// Accept `128MiB`-style sizes on the wire while keeping the field a plain
18/// `u64`. `ByteSize` is a parsing convenience, not part of the batching API.
19fn de_byte_size<'de, D: Deserializer<'de>>(d: D) -> Result<u64, D::Error> {
20 ByteSize::deserialize(d).map(|b| b.as_u64())
21}
22
23/// Batch sealing thresholds for one shard worker. A batch seals as soon as
24/// **any** threshold trips; since chunks arrive whole, a sealed batch may
25/// overshoot `max_rows`/`max_bytes` by at most one chunk.
26#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
27#[serde(deny_unknown_fields, default)]
28#[non_exhaustive]
29pub struct BatchConfig {
30 /// Seal at this many rows (for a message-oriented sink, messages).
31 pub max_rows: u64,
32 /// Seal at this many encoded, uncompressed bytes.
33 #[serde(deserialize_with = "de_byte_size")]
34 pub max_bytes: u64,
35 /// Seal a non-empty batch this long after its first chunk arrived,
36 /// bounding latency at low throughput.
37 #[serde(with = "humantime_serde")]
38 pub linger: Duration,
39}
40
41impl Default for BatchConfig {
42 fn default() -> Self {
43 BatchConfig {
44 max_rows: 500_000,
45 max_bytes: 128 * 1024 * 1024,
46 linger: Duration::from_secs(1),
47 }
48 }
49}
50
51/// In-flight write limits for one shard worker.
52#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
53#[serde(deny_unknown_fields, default)]
54#[non_exhaustive]
55pub struct InflightConfig {
56 /// Concurrent sealed batches per shard (for a replicated sink, writes to
57 /// different replicas). While all permits are taken the worker stops
58 /// consuming its queue, which fills and surfaces as backpressure.
59 pub max_per_shard: usize,
60}
61
62impl Default for InflightConfig {
63 fn default() -> Self {
64 InflightConfig { max_per_shard: 2 }
65 }
66}
67
68/// Retry policy for batch writes. Retries rotate across healthy replicas;
69/// the sealed batch and its deduplication token are reused unchanged.
70#[derive(Clone, Copy, Debug, PartialEq, Deserialize)]
71#[serde(deny_unknown_fields, default)]
72#[non_exhaustive]
73pub struct RetryConfig {
74 /// First backoff delay.
75 #[serde(with = "humantime_serde")]
76 pub initial: Duration,
77 /// Backoff cap.
78 #[serde(with = "humantime_serde")]
79 pub max: Duration,
80 /// Backoff growth factor per attempt.
81 pub multiplier: f64,
82 /// Fraction of the delay randomized away (`0.0..=1.0`).
83 pub jitter: f64,
84 /// Total write attempts before the batch is abandoned (acknowledgments
85 /// failed, watermark stalls). `0` means unbounded, retrying until the drain
86 /// deadline, at which point the attempt in flight is aborted and the batch
87 /// abandoned. The at-least-once default.
88 ///
89 /// An unbounded policy holds its in-flight slot
90 /// ([`InflightConfig::max_per_shard`]) for the whole outage, since a slot
91 /// frees only when its write task ends. That is how a down sink
92 /// back-pressures the source rather than buffering, and it means a shard
93 /// talking to a dead sink runs at zero in-flight capacity until either the
94 /// sink recovers or the drain deadline arrives.
95 pub max_attempts: u32,
96}
97
98impl Default for RetryConfig {
99 fn default() -> Self {
100 RetryConfig {
101 initial: Duration::from_millis(100),
102 max: Duration::from_secs(10),
103 multiplier: 2.0,
104 jitter: 0.2,
105 max_attempts: 0,
106 }
107 }
108}
109
110/// Why a [`RetryConfig`] was rejected.
111///
112/// Each message names the offending key relative to the sink's `retry`
113/// section; connectors prepend their own config path when converting it into
114/// their `ConfigError`.
115#[derive(Clone, Debug, PartialEq, thiserror::Error)]
116#[non_exhaustive]
117pub enum RetryConfigError {
118 /// `multiplier` is not a finite number in `[1.0, 1e9]`. Below `1.0` the
119 /// delay shrinks instead of backing off.
120 #[error("retry.multiplier must be a finite number in [1.0, 1e9] (got {0})")]
121 Multiplier(f64),
122 /// `jitter` is not a finite fraction in `[0.0, 1.0]`.
123 #[error("retry.jitter must be a finite fraction in [0.0, 1.0] (got {0})")]
124 Jitter(f64),
125 /// `initial` or `max` is zero, leaving no delay to sleep at all.
126 #[error("retry.initial and retry.max must be non-zero")]
127 ZeroDelay,
128 /// `initial` is larger than the ceiling it grows towards.
129 #[error("retry.initial ({initial:?}) must not exceed retry.max ({max:?})")]
130 InitialExceedsMax {
131 /// The configured first delay.
132 initial: Duration,
133 /// The configured ceiling.
134 max: Duration,
135 },
136}
137
138impl RetryConfig {
139 /// Reject a retry policy that would misbehave at runtime. Connectors call
140 /// this from their config validation and prepend their own config path to
141 /// the message, so the rules stay in one place instead of being mirrored
142 /// per connector.
143 ///
144 /// This is about intent, not safety. `Backoff` never
145 /// panics for *any* `RetryConfig` and always saturates at `max`. It
146 /// returns a zero delay only for a policy this rejects (`initial` or
147 /// `max` of zero), so "never zero" is a property of a **validated**
148 /// policy, not of the type. What it catches is a policy no operator
149 /// means: a sub-`1.0` multiplier shrinks the delay instead of backing
150 /// off, a zero delay is not a backoff at all, and both are worth failing
151 /// at load rather than at 3am.
152 ///
153 /// The bounds are generous and do **not** guarantee a *sensible* policy.
154 /// `initial: 1ns, max: 1ns` passes. They rule out the nonsensical, not
155 /// the aggressive.
156 ///
157 /// # Errors
158 ///
159 /// [`RetryConfigError`], naming the offending key.
160 ///
161 /// ```
162 /// use spate_core::sink::{RetryConfig, RetryConfigError};
163 ///
164 /// assert!(RetryConfig::default().validate().is_ok());
165 ///
166 /// let mut hot_loop = RetryConfig::default();
167 /// hot_loop.multiplier = 0.5;
168 /// assert_eq!(hot_loop.validate(), Err(RetryConfigError::Multiplier(0.5)));
169 /// ```
170 pub fn validate(&self) -> Result<(), RetryConfigError> {
171 if !self.multiplier.is_finite() || !(1.0..=1e9).contains(&self.multiplier) {
172 return Err(RetryConfigError::Multiplier(self.multiplier));
173 }
174 if !self.jitter.is_finite() || !(0.0..=1.0).contains(&self.jitter) {
175 return Err(RetryConfigError::Jitter(self.jitter));
176 }
177 if self.initial.is_zero() || self.max.is_zero() {
178 return Err(RetryConfigError::ZeroDelay);
179 }
180 if self.initial > self.max {
181 return Err(RetryConfigError::InitialExceedsMax {
182 initial: self.initial,
183 max: self.max,
184 });
185 }
186 Ok(())
187 }
188
189 /// Whether this policy lets a shard sleep indefinitely without ever
190 /// giving up on the batch.
191 ///
192 /// It takes *both* halves: unbounded attempts, so the batch is never
193 /// abandoned, and a ceiling long enough that a sleeping shard is
194 /// indistinguishable from a wedged one. With a finite `max_attempts` the
195 /// batch is abandoned and the stall is bounded; with a short ceiling the
196 /// retries keep visibly ticking. Only the combination goes quiet.
197 ///
198 /// Not a [`validate`](Self::validate) rule. Nothing here is unsafe; the
199 /// drain deadline still aborts the sleep at shutdown, so at-least-once
200 /// holds. A sink fronting an expensive or rate-limited destination may
201 /// well mean it. The threshold is a heuristic, which a warning may use
202 /// and a rejection may not.
203 pub(crate) fn stalls_indefinitely(&self) -> bool {
204 /// Past this, an unbounded policy stops reading as a backoff.
205 const CEILING: Duration = Duration::from_secs(300);
206 self.max_attempts == 0 && self.max > CEILING
207 }
208}
209
210/// Per-replica circuit breaker thresholds.
211#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
212#[serde(deny_unknown_fields, default)]
213#[non_exhaustive]
214pub struct BreakerConfig {
215 /// Consecutive failures that open the breaker, quarantining that endpoint.
216 pub failure_threshold: u32,
217 /// How long an open breaker rejects a replica before probing again.
218 ///
219 /// Also bounds how long a batch parked behind a fully-probing shard waits
220 /// before re-checking of its own accord (clamped to `[100ms, 30s]`).
221 /// Capped at a year on the way in: it is stamped into a deadline, and
222 /// `Instant + Duration` panics rather than saturating.
223 #[serde(with = "humantime_serde")]
224 pub open_for: Duration,
225 /// Concurrent probe writes allowed while half-open.
226 ///
227 /// Must be at least 1; [`validate`](Self::validate) rejects `0`, which
228 /// taken literally would mean the replica never recovers. The breaker also
229 /// floors it at 1 at the point of use, so a config built programmatically
230 /// rather than loaded cannot wedge a replica either.
231 pub half_open_probes: u32,
232}
233
234impl Default for BreakerConfig {
235 fn default() -> Self {
236 BreakerConfig {
237 failure_threshold: 3,
238 open_for: Duration::from_secs(5),
239 half_open_probes: 1,
240 }
241 }
242}
243
244impl BreakerConfig {
245 /// Ceiling on [`open_for`](Self::open_for).
246 ///
247 /// It is stamped into a deadline, and `Instant + Duration` panics rather
248 /// than saturating. A year already means "never probe again", so anything
249 /// beyond it is a typo rather than a policy.
250 pub const MAX_OPEN_FOR: Duration = Duration::from_secs(365 * 24 * 60 * 60);
251
252 /// Reject breaker thresholds that would misbehave at runtime.
253 ///
254 /// The companion to [`RetryConfig::validate`], called from the same place
255 /// for the same reason: the rules stay here instead of being mirrored per
256 /// connector.
257 ///
258 /// # Errors
259 ///
260 /// [`BreakerConfigError`], naming the offending key.
261 ///
262 /// ```
263 /// use spate_core::sink::{BreakerConfig, BreakerConfigError};
264 ///
265 /// assert!(BreakerConfig::default().validate().is_ok());
266 ///
267 /// let mut wedged = BreakerConfig::default();
268 /// wedged.half_open_probes = 0;
269 /// assert_eq!(wedged.validate(), Err(BreakerConfigError::ZeroHalfOpenProbes));
270 /// ```
271 pub fn validate(&self) -> Result<(), BreakerConfigError> {
272 if self.half_open_probes == 0 {
273 return Err(BreakerConfigError::ZeroHalfOpenProbes);
274 }
275 if self.open_for.is_zero() {
276 return Err(BreakerConfigError::ZeroOpenFor);
277 }
278 if self.open_for > Self::MAX_OPEN_FOR {
279 return Err(BreakerConfigError::OpenForTooLong(self.open_for));
280 }
281 Ok(())
282 }
283}
284
285/// Why a [`BreakerConfig`] was rejected.
286#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
287#[non_exhaustive]
288pub enum BreakerConfigError {
289 /// `half_open_probes` is zero, so no probe budget exists to recover with.
290 #[error("breaker.half_open_probes must be at least 1")]
291 ZeroHalfOpenProbes,
292 /// `open_for` is zero, which is not a quarantine at all.
293 #[error("breaker.open_for must be non-zero")]
294 ZeroOpenFor,
295 /// `open_for` exceeds [`BreakerConfig::MAX_OPEN_FOR`].
296 #[error("breaker.open_for must not exceed a year (got {0:?})")]
297 OpenForTooLong(Duration),
298}
299
300/// Complete sink worker-pool configuration.
301#[derive(Clone, Copy, Debug, Default, PartialEq)]
302#[non_exhaustive]
303pub struct SinkPoolConfig {
304 /// Batch sealing thresholds.
305 pub batch: BatchConfig,
306 /// In-flight limits.
307 pub inflight: InflightConfig,
308 /// Write retry policy.
309 pub retry: RetryConfig,
310 /// Replica circuit breaker.
311 pub breaker: BreakerConfig,
312}
313
314impl SinkPoolConfig {
315 /// All four sections at once, as a connector's factory assembles them
316 /// from its config.
317 ///
318 /// ```
319 /// use spate_core::sink::{BatchConfig, BreakerConfig, InflightConfig, RetryConfig, SinkPoolConfig};
320 ///
321 /// let mut batch = BatchConfig::default();
322 /// batch.max_rows = 1_000;
323 /// let pool = SinkPoolConfig::new(
324 /// batch,
325 /// InflightConfig::default(),
326 /// RetryConfig::default(),
327 /// BreakerConfig::default(),
328 /// );
329 /// assert_eq!(pool.batch.max_rows, 1_000);
330 /// ```
331 #[must_use]
332 pub fn new(
333 batch: BatchConfig,
334 inflight: InflightConfig,
335 retry: RetryConfig,
336 breaker: BreakerConfig,
337 ) -> SinkPoolConfig {
338 SinkPoolConfig {
339 batch,
340 inflight,
341 retry,
342 breaker,
343 }
344 }
345}
346
347#[cfg(test)]
348mod tests {
349 use super::*;
350
351 fn retry(mutate: impl FnOnce(&mut RetryConfig)) -> RetryConfig {
352 let mut cfg = RetryConfig::default();
353 mutate(&mut cfg);
354 cfg
355 }
356
357 #[test]
358 fn validate_rejects_policies_that_misbehave_at_runtime() {
359 let cases = [
360 (retry(|c| c.multiplier = 0.5), "multiplier"),
361 (retry(|c| c.multiplier = -2.0), "multiplier"),
362 (retry(|c| c.multiplier = f64::NAN), "multiplier"),
363 (retry(|c| c.multiplier = f64::INFINITY), "multiplier"),
364 (retry(|c| c.multiplier = 1e9 + 1.0), "multiplier"),
365 (retry(|c| c.jitter = 1.5), "jitter"),
366 (retry(|c| c.jitter = -0.1), "jitter"),
367 (retry(|c| c.jitter = f64::NAN), "jitter"),
368 (retry(|c| c.initial = Duration::ZERO), "non-zero"),
369 (retry(|c| c.max = Duration::ZERO), "non-zero"),
370 (
371 retry(|c| {
372 c.initial = Duration::from_secs(10);
373 c.max = Duration::from_secs(1);
374 }),
375 "must not exceed",
376 ),
377 ];
378 for (cfg, needle) in cases {
379 let err = cfg
380 .validate()
381 .expect_err(&format!("{cfg:?} must fail"))
382 .to_string();
383 assert!(err.contains(needle), "expected `{needle}` in `{err}`");
384 }
385 }
386
387 #[test]
388 fn the_error_messages_survived_the_move_into_the_framework() {
389 // Both sinks mirrored these strings before the rules moved here, and
390 // their config tests still assert on them under their own prefix.
391 assert_eq!(
392 RetryConfigError::Multiplier(0.5).to_string(),
393 "retry.multiplier must be a finite number in [1.0, 1e9] (got 0.5)"
394 );
395 assert_eq!(
396 RetryConfigError::Jitter(1.5).to_string(),
397 "retry.jitter must be a finite fraction in [0.0, 1.0] (got 1.5)"
398 );
399 assert_eq!(
400 RetryConfigError::ZeroDelay.to_string(),
401 "retry.initial and retry.max must be non-zero"
402 );
403 assert_eq!(
404 RetryConfigError::InitialExceedsMax {
405 initial: Duration::from_secs(10),
406 max: Duration::from_secs(1),
407 }
408 .to_string(),
409 "retry.initial (10s) must not exceed retry.max (1s)"
410 );
411 }
412
413 #[test]
414 fn only_unbounded_attempts_with_a_long_ceiling_count_as_a_stall() {
415 let long = Duration::from_secs(3600);
416 // Both halves, the only combination that goes quiet.
417 assert!(retry(|c| c.max = long).stalls_indefinitely());
418 // A finite attempt cap abandons the batch; the stall is bounded.
419 assert!(
420 !retry(|c| {
421 c.max = long;
422 c.max_attempts = 5;
423 })
424 .stalls_indefinitely()
425 );
426 // A short ceiling keeps the retries visibly ticking.
427 assert!(!RetryConfig::default().stalls_indefinitely());
428 // Right at the threshold is still fine; it is an upper bound.
429 assert!(!retry(|c| c.max = Duration::from_secs(300)).stalls_indefinitely());
430 assert!(retry(|c| c.max = Duration::from_secs(301)).stalls_indefinitely());
431 }
432
433 #[test]
434 fn validate_accepts_the_default_and_the_boundaries() {
435 assert!(RetryConfig::default().validate().is_ok());
436 let boundary = RetryConfig {
437 initial: Duration::from_nanos(1),
438 max: Duration::from_nanos(1),
439 multiplier: 1.0,
440 jitter: 0.0,
441 max_attempts: 0,
442 };
443 assert!(boundary.validate().is_ok(), "{boundary:?}");
444 let upper = RetryConfig {
445 multiplier: 1e9,
446 jitter: 1.0,
447 ..RetryConfig::default()
448 };
449 assert!(upper.validate().is_ok(), "{upper:?}");
450 }
451
452 fn breaker(mutate: impl FnOnce(&mut BreakerConfig)) -> BreakerConfig {
453 let mut cfg = BreakerConfig::default();
454 mutate(&mut cfg);
455 cfg
456 }
457
458 #[test]
459 fn breaker_validate_rejects_a_budget_no_replica_could_recover_from() {
460 assert_eq!(
461 breaker(|c| c.half_open_probes = 0).validate(),
462 Err(BreakerConfigError::ZeroHalfOpenProbes)
463 );
464 assert_eq!(
465 breaker(|c| c.open_for = Duration::ZERO).validate(),
466 Err(BreakerConfigError::ZeroOpenFor)
467 );
468 let too_long = BreakerConfig::MAX_OPEN_FOR + Duration::from_secs(1);
469 assert_eq!(
470 breaker(|c| c.open_for = too_long).validate(),
471 Err(BreakerConfigError::OpenForTooLong(too_long))
472 );
473 }
474
475 #[test]
476 fn breaker_validate_accepts_the_default_and_the_boundaries() {
477 assert!(BreakerConfig::default().validate().is_ok());
478 // Both ends of every bound, so tightening one shows up here rather
479 // than in a connector's load path.
480 assert!(breaker(|c| c.half_open_probes = 1).validate().is_ok());
481 assert!(
482 breaker(|c| c.open_for = Duration::from_nanos(1))
483 .validate()
484 .is_ok()
485 );
486 assert!(
487 breaker(|c| c.open_for = BreakerConfig::MAX_OPEN_FOR)
488 .validate()
489 .is_ok()
490 );
491 }
492
493 /// The cap exists because `on_failure` stamps `now + open_for` and
494 /// `Instant + Duration` panics on overflow. Validation rejects the value,
495 /// but `BreakerConfig` is a public `Copy` struct, so the breaker floors and
496 /// caps at the point of use too. This pins that the two agree on where
497 /// the line is.
498 #[test]
499 fn breaker_cap_matches_what_the_breaker_applies() {
500 assert_eq!(
501 BreakerConfig::MAX_OPEN_FOR,
502 Duration::from_secs(365 * 24 * 60 * 60)
503 );
504 }
505}