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
8/// Start the periodic same-stream cold chunk compactor when explicitly enabled.
9pub fn spawn_cold_compaction_worker_if_configured(
10    runtime: &ShardRuntime,
11    config: &ursula_config::ColdConfig,
12) {
13    if !config.compaction_enabled {
14        return;
15    }
16    let interval = config.compaction_interval.as_duration();
17    let target_bytes = config.compaction_target_size.as_bytes();
18    let max_bytes = config.compaction_max_size.as_bytes();
19    let max_streams = config.compaction_max_streams_per_pass.max(1);
20    let gc_grace_ms =
21        u64::try_from(config.compaction_gc_grace.as_duration().as_millis()).unwrap_or(u64::MAX);
22    let runtime = runtime.clone();
23    tokio::spawn(async move {
24        loop {
25            match runtime
26                .compact_cold_once(target_bytes, max_bytes, max_streams, gc_grace_ms)
27                .await
28            {
29                Ok(compacted) if compacted > 0 => {
30                    tracing::info!(compacted, "cold chunk compaction pass completed");
31                }
32                Ok(_) => {}
33                Err(err) => tracing::error!("cold compaction worker error: {err}"),
34            }
35            tokio::time::sleep(interval).await;
36        }
37    });
38}
39
40/// Start the periodic cold-flush worker if the configured interval is non-zero.
41pub fn spawn_cold_flush_worker_if_configured(
42    runtime: &ShardRuntime,
43    config: &ursula_config::ColdConfig,
44) {
45    let interval = config.flush_interval.as_duration();
46    if interval.is_zero() {
47        return;
48    }
49    let min_hot_bytes = usize::try_from(config.flush_min_hot_size().as_bytes())
50        .expect("config validation ensures flush sizes fit usize");
51    let max_flush_bytes = usize::try_from(config.flush_max_size().as_bytes())
52        .expect("config validation ensures flush sizes fit usize");
53    let max_concurrency = config.flush_max_concurrency.max(1);
54    let runtime = runtime.clone();
55    tokio::spawn(async move {
56        loop {
57            if let Err(err) = runtime
58                .flush_cold_all_groups_once_bounded(
59                    PlanGroupColdFlushRequest {
60                        min_hot_bytes,
61                        max_flush_bytes,
62                        max_batch_bytes: max_flush_bytes,
63                    },
64                    max_concurrency,
65                )
66                .await
67            {
68                tracing::error!("cold flush worker error: {err}");
69            }
70            tokio::time::sleep(interval).await;
71        }
72    });
73}
74
75/// Start the periodic cold-gc worker if the configured interval is non-zero.
76pub fn spawn_cold_gc_worker_if_configured(
77    runtime: &ShardRuntime,
78    config: &ursula_config::ColdConfig,
79) {
80    let interval = config.gc_interval.as_duration();
81    if interval.is_zero() {
82        return;
83    }
84    let max_entries = config.gc_max_entries.max(1);
85    let runtime = runtime.clone();
86    tokio::spawn(async move {
87        loop {
88            if let Err(err) = runtime.run_cold_gc_all_groups_once(max_entries).await {
89                tracing::error!("cold gc worker error: {err}");
90            }
91            tokio::time::sleep(interval).await;
92        }
93    });
94}
ursula_runtime/cold_worker.rs

ursula_runtime/
cold_worker.rs
