ursula_runtime/
cold_worker.rs

1//! Cold-tier background workers.
2//!
3//! Started by the bootstrap layer after the runtime is constructed.
4
5use crate::PlanGroupColdFlushRequest;
6use crate::ShardRuntime;
7
8fn effective_min_hot_bytes(
9    normal_min_hot_bytes: usize,
10    observed_hot_bytes: u64,
11    pressure_hot_bytes: u64,
12) -> (usize, bool) {
13    let pressure_active = pressure_hot_bytes > 0 && observed_hot_bytes >= pressure_hot_bytes;
14    (
15        if pressure_active {
16            1
17        } else {
18            normal_min_hot_bytes
19        },
20        pressure_active,
21    )
22}
23
24/// Start the periodic same-stream cold chunk compactor when explicitly enabled.
25pub fn spawn_cold_compaction_worker_if_configured(
26    runtime: &ShardRuntime,
27    config: &ursula_config::ColdConfig,
28) {
29    if !config.compaction_enabled {
30        return;
31    }
32    let interval = config.compaction_interval.as_duration();
33    let target_bytes = config.compaction_target_size.as_bytes();
34    let max_bytes = config.compaction_max_size.as_bytes();
35    let max_streams = config.compaction_max_streams_per_pass.max(1);
36    let gc_grace_ms =
37        u64::try_from(config.compaction_gc_grace.as_duration().as_millis()).unwrap_or(u64::MAX);
38    let runtime = runtime.clone();
39    tokio::spawn(async move {
40        loop {
41            match runtime
42                .compact_cold_once(target_bytes, max_bytes, max_streams, gc_grace_ms)
43                .await
44            {
45                Ok(compacted) if compacted > 0 => {
46                    tracing::info!(compacted, "cold chunk compaction pass completed");
47                }
48                Ok(_) => {}
49                Err(err) => tracing::error!("cold compaction worker error: {err}"),
50            }
51            tokio::time::sleep(interval).await;
52        }
53    });
54}
55
56/// Start the periodic cold-flush worker if the configured interval is non-zero.
57pub fn spawn_cold_flush_worker_if_configured(
58    runtime: &ShardRuntime,
59    config: &ursula_config::ColdConfig,
60) {
61    let interval = config.flush_interval.as_duration();
62    if interval.is_zero() {
63        return;
64    }
65    let min_hot_bytes = usize::try_from(config.flush_min_hot_size().as_bytes())
66        .expect("config validation ensures flush sizes fit usize");
67    let max_flush_bytes = usize::try_from(config.flush_max_size().as_bytes())
68        .expect("config validation ensures flush sizes fit usize");
69    let pressure_hot_bytes = config.flush_pressure_hot_size.as_bytes();
70    let max_concurrency = config.flush_max_concurrency.max(1);
71    let runtime = runtime.clone();
72    tokio::spawn(async move {
73        loop {
74            let metrics = runtime.metrics();
75            let observed_hot_bytes = metrics.inner.cold_hot_bytes();
76            let (pass_min_hot_bytes, pressure_active) =
77                effective_min_hot_bytes(min_hot_bytes, observed_hot_bytes, pressure_hot_bytes);
78            match runtime
79                .flush_cold_all_groups_once_bounded(
80                    PlanGroupColdFlushRequest {
81                        min_hot_bytes: pass_min_hot_bytes,
82                        max_flush_bytes,
83                        max_batch_bytes: max_flush_bytes,
84                    },
85                    max_concurrency,
86                )
87                .await
88            {
89                Ok(flushed) if pressure_active => {
90                    metrics.inner.record_cold_pressure_flush(flushed);
91                    if flushed > 0 {
92                        tracing::info!(
93                            observed_hot_bytes,
94                            pressure_hot_bytes,
95                            flushed,
96                            "cold pressure flush pass completed"
97                        );
98                    }
99                }
100                Ok(_) => {}
101                Err(err) => tracing::error!("cold flush worker error: {err}"),
102            }
103            tokio::time::sleep(interval).await;
104        }
105    });
106}
107
108/// Start the periodic cold-gc worker if the configured interval is non-zero.
109pub fn spawn_cold_gc_worker_if_configured(
110    runtime: &ShardRuntime,
111    config: &ursula_config::ColdConfig,
112) {
113    let interval = config.gc_interval.as_duration();
114    if interval.is_zero() {
115        return;
116    }
117    let max_entries = config.gc_max_entries.max(1);
118    let runtime = runtime.clone();
119    tokio::spawn(async move {
120        loop {
121            if let Err(err) = runtime.run_cold_gc_all_groups_once(max_entries).await {
122                tracing::error!("cold gc worker error: {err}");
123            }
124            tokio::time::sleep(interval).await;
125        }
126    });
127}
128
129#[cfg(test)]
130mod tests {
131    use super::effective_min_hot_bytes;
132
133    #[test]
134    fn pressure_flush_activates_at_the_aggregate_watermark() {
135        assert_eq!(effective_min_hot_bytes(8, 127, 128), (8, false));
136        assert_eq!(effective_min_hot_bytes(8, 128, 128), (1, true));
137        assert_eq!(effective_min_hot_bytes(8, 129, 128), (1, true));
138    }
139
140    #[test]
141    fn zero_pressure_watermark_disables_the_fallback() {
142        assert_eq!(effective_min_hot_bytes(8, u64::MAX, 0), (8, false));
143    }
144}
ursula_runtime/cold_worker.rs

ursula_runtime/
cold_worker.rs
