etl_core/sink/
config.rs

1//! Sink worker-pool tuning knobs.
2//!
3//! These are framework-level structs; the pipeline runtime maps the user's
4//! YAML sink section onto them.
5
6use std::time::Duration;
7
8/// Batch sealing thresholds for one shard worker. A batch seals as soon as
9/// **any** threshold trips; since chunks arrive whole, a sealed batch may
10/// overshoot `max_rows`/`max_bytes` by at most one chunk.
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
12pub struct BatchConfig {
13    /// Seal at this many rows.
14    pub max_rows: u64,
15    /// Seal at this many encoded bytes.
16    pub max_bytes: u64,
17    /// Seal a non-empty batch this long after its first chunk arrived,
18    /// bounding latency at low throughput.
19    pub linger: Duration,
20}
21
22impl Default for BatchConfig {
23    fn default() -> Self {
24        BatchConfig {
25            max_rows: 500_000,
26            max_bytes: 128 * 1024 * 1024,
27            linger: Duration::from_secs(1),
28        }
29    }
30}
31
32/// In-flight write limits for one shard worker.
33#[derive(Clone, Copy, Debug, PartialEq, Eq)]
34pub struct InflightConfig {
35    /// Concurrent sealed batches per shard. While all permits are taken the
36    /// worker stops consuming its queue, which fills and surfaces as
37    /// backpressure.
38    pub max_per_shard: usize,
39}
40
41impl Default for InflightConfig {
42    fn default() -> Self {
43        InflightConfig { max_per_shard: 2 }
44    }
45}
46
47/// Retry policy for batch writes. Retries rotate across healthy replicas;
48/// the sealed batch and its deduplication token are reused unchanged.
49#[derive(Clone, Copy, Debug, PartialEq)]
50pub struct RetryConfig {
51    /// First backoff delay.
52    pub initial: Duration,
53    /// Backoff cap.
54    pub max: Duration,
55    /// Backoff growth factor per attempt.
56    pub multiplier: f64,
57    /// Fraction of the delay randomized away (`0.0..=1.0`).
58    pub jitter: f64,
59    /// Total write attempts before the batch is abandoned (acknowledgements
60    /// failed, watermark stalls). `0` means unbounded — retry until the
61    /// drain deadline, the at-least-once default.
62    pub max_attempts: u32,
63}
64
65impl Default for RetryConfig {
66    fn default() -> Self {
67        RetryConfig {
68            initial: Duration::from_millis(100),
69            max: Duration::from_secs(10),
70            multiplier: 2.0,
71            jitter: 0.2,
72            max_attempts: 0,
73        }
74    }
75}
76
77/// Per-replica circuit breaker thresholds.
78#[derive(Clone, Copy, Debug, PartialEq, Eq)]
79pub struct BreakerConfig {
80    /// Consecutive failures that open the breaker.
81    pub failure_threshold: u32,
82    /// How long an open breaker rejects a replica before probing again.
83    pub open_for: Duration,
84    /// Concurrent probe writes allowed while half-open.
85    pub half_open_probes: u32,
86}
87
88impl Default for BreakerConfig {
89    fn default() -> Self {
90        BreakerConfig {
91            failure_threshold: 3,
92            open_for: Duration::from_secs(5),
93            half_open_probes: 1,
94        }
95    }
96}
97
98/// Complete sink worker-pool configuration.
99#[derive(Clone, Copy, Debug, Default, PartialEq)]
100pub struct SinkPoolConfig {
101    /// Batch sealing thresholds.
102    pub batch: BatchConfig,
103    /// In-flight limits.
104    pub inflight: InflightConfig,
105    /// Write retry policy.
106    pub retry: RetryConfig,
107    /// Replica circuit breaker.
108    pub breaker: BreakerConfig,
109}
etl_core/sink/config.rs

etl_core/sink/
config.rs
