1use std::str::FromStr;
2use std::time::Duration;
3
4use thiserror::Error;
5
6const DEFAULT_POLL_INTERVAL_MS: u64 = 500;
7const DEFAULT_CLAIM_BATCH_SIZE: i64 = 16;
8const DEFAULT_LEASE_TTL_SECONDS: i32 = 60;
9const DEFAULT_MAX_GLOBAL_CONCURRENCY: usize = 32;
10const DEFAULT_REAPER_INTERVAL_SECONDS: u64 = 15;
11const DEFAULT_SCHEDULE_POLL_INTERVAL_SECONDS: u64 = 30;
12const DEFAULT_REAPER_RETRY_DELAY_MS: i32 = 30_000;
13
14#[derive(Debug, Clone)]
15pub struct JobsConfig {
16 pub worker_id: String,
17 pub poll_interval: Duration,
18 pub claim_batch_size: i64,
19 pub lease_ttl_seconds: i32,
20 pub max_global_concurrency: usize,
21 pub reaper_interval: Duration,
22 pub schedule_poll_interval: Duration,
23 pub reaper_retry_delay_ms: i32,
24}
25
26#[non_exhaustive]
27#[derive(Debug, Clone, Copy, Error, Eq, PartialEq)]
28pub enum JobsConfigValidationError {
29 #[error("jobs config worker_id must not be empty")]
30 EmptyWorkerId,
31 #[error("jobs config poll_interval must be greater than zero")]
32 ZeroPollInterval,
33 #[error("jobs config claim_batch_size must be at least 1, got {actual}")]
34 InvalidClaimBatchSize { actual: i64 },
35 #[error("jobs config lease_ttl_seconds must be at least 1, got {actual}")]
36 InvalidLeaseTtlSeconds { actual: i32 },
37 #[error("jobs config max_global_concurrency must be at least 1")]
38 InvalidMaxGlobalConcurrency,
39 #[error("jobs config reaper_interval must be greater than zero")]
40 ZeroReaperInterval,
41 #[error("jobs config schedule_poll_interval must be greater than zero")]
42 ZeroSchedulePollInterval,
43 #[error("jobs config reaper_retry_delay_ms must be at least 1, got {actual}")]
44 InvalidReaperRetryDelayMs { actual: i32 },
45}
46
47impl JobsConfig {
48 #[must_use]
49 pub fn from_env() -> Self {
50 Self {
51 worker_id: std::env::var("JOBS_WORKER_ID")
52 .ok()
53 .filter(|value| !value.trim().is_empty())
54 .unwrap_or_else(|| format!("worker-{}", uuid::Uuid::now_v7())),
55 poll_interval: Duration::from_millis(
56 parse_env("JOBS_POLL_INTERVAL_MS", DEFAULT_POLL_INTERVAL_MS).max(1),
57 ),
58 claim_batch_size: parse_env("JOBS_CLAIM_BATCH_SIZE", DEFAULT_CLAIM_BATCH_SIZE).max(1),
59 lease_ttl_seconds: parse_env("JOBS_LEASE_TTL_SECONDS", DEFAULT_LEASE_TTL_SECONDS)
60 .max(10),
61 max_global_concurrency: parse_env(
62 "JOBS_MAX_GLOBAL_CONCURRENCY",
63 DEFAULT_MAX_GLOBAL_CONCURRENCY,
64 )
65 .max(1),
66 reaper_interval: Duration::from_secs(
67 parse_env(
68 "JOBS_REAPER_INTERVAL_SECONDS",
69 DEFAULT_REAPER_INTERVAL_SECONDS,
70 )
71 .max(1),
72 ),
73 schedule_poll_interval: Duration::from_secs(
74 parse_env(
75 "JOBS_SCHEDULE_POLL_INTERVAL_SECONDS",
76 DEFAULT_SCHEDULE_POLL_INTERVAL_SECONDS,
77 )
78 .max(1),
79 ),
80 reaper_retry_delay_ms: parse_env(
81 "JOBS_REAPER_RETRY_DELAY_MS",
82 DEFAULT_REAPER_RETRY_DELAY_MS,
83 )
84 .max(1_000),
85 }
86 }
87
88 pub fn validate(&self) -> Result<(), JobsConfigValidationError> {
89 if self.worker_id.trim().is_empty() {
90 return Err(JobsConfigValidationError::EmptyWorkerId);
91 }
92 if self.poll_interval.is_zero() {
93 return Err(JobsConfigValidationError::ZeroPollInterval);
94 }
95 if self.claim_batch_size < 1 {
96 return Err(JobsConfigValidationError::InvalidClaimBatchSize {
97 actual: self.claim_batch_size,
98 });
99 }
100 if self.lease_ttl_seconds < 1 {
101 return Err(JobsConfigValidationError::InvalidLeaseTtlSeconds {
102 actual: self.lease_ttl_seconds,
103 });
104 }
105 if self.max_global_concurrency < 1 {
106 return Err(JobsConfigValidationError::InvalidMaxGlobalConcurrency);
107 }
108 if self.reaper_interval.is_zero() {
109 return Err(JobsConfigValidationError::ZeroReaperInterval);
110 }
111 if self.schedule_poll_interval.is_zero() {
112 return Err(JobsConfigValidationError::ZeroSchedulePollInterval);
113 }
114 if self.reaper_retry_delay_ms < 1 {
115 return Err(JobsConfigValidationError::InvalidReaperRetryDelayMs {
116 actual: self.reaper_retry_delay_ms,
117 });
118 }
119
120 Ok(())
121 }
122
123 pub(crate) fn validate_worker_loop(&self) -> Result<(), JobsConfigValidationError> {
124 if self.worker_id.trim().is_empty() {
125 return Err(JobsConfigValidationError::EmptyWorkerId);
126 }
127 if self.poll_interval.is_zero() {
128 return Err(JobsConfigValidationError::ZeroPollInterval);
129 }
130 if self.claim_batch_size < 1 {
131 return Err(JobsConfigValidationError::InvalidClaimBatchSize {
132 actual: self.claim_batch_size,
133 });
134 }
135 if self.lease_ttl_seconds < 1 {
136 return Err(JobsConfigValidationError::InvalidLeaseTtlSeconds {
137 actual: self.lease_ttl_seconds,
138 });
139 }
140 if self.max_global_concurrency < 1 {
141 return Err(JobsConfigValidationError::InvalidMaxGlobalConcurrency);
142 }
143
144 Ok(())
145 }
146
147 pub(crate) fn validate_scheduler_loop(&self) -> Result<(), JobsConfigValidationError> {
148 if self.claim_batch_size < 1 {
149 return Err(JobsConfigValidationError::InvalidClaimBatchSize {
150 actual: self.claim_batch_size,
151 });
152 }
153 if self.schedule_poll_interval.is_zero() {
154 return Err(JobsConfigValidationError::ZeroSchedulePollInterval);
155 }
156
157 Ok(())
158 }
159
160 pub(crate) fn validate_reaper_loop(&self) -> Result<(), JobsConfigValidationError> {
161 if self.claim_batch_size < 1 {
162 return Err(JobsConfigValidationError::InvalidClaimBatchSize {
163 actual: self.claim_batch_size,
164 });
165 }
166 if self.reaper_interval.is_zero() {
167 return Err(JobsConfigValidationError::ZeroReaperInterval);
168 }
169 if self.reaper_retry_delay_ms < 1 {
170 return Err(JobsConfigValidationError::InvalidReaperRetryDelayMs {
171 actual: self.reaper_retry_delay_ms,
172 });
173 }
174
175 Ok(())
176 }
177}
178
179fn parse_env<T>(name: &str, default: T) -> T
180where
181 T: FromStr,
182{
183 std::env::var(name)
184 .ok()
185 .and_then(|value| value.parse::<T>().ok())
186 .unwrap_or(default)
187}
188
189#[cfg(test)]
190mod tests {
191 use std::sync::{Mutex, OnceLock};
192
193 use super::*;
194
195 static ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
196
197 #[derive(Debug)]
198 struct ScopedEnv {
199 _guard: std::sync::MutexGuard<'static, ()>,
200 prior: Vec<(String, Option<String>)>,
201 }
202
203 impl ScopedEnv {
204 fn set(overrides: &[(&str, Option<&str>)]) -> Self {
205 let guard = ENV_LOCK
206 .get_or_init(|| Mutex::new(()))
207 .lock()
208 .unwrap_or_else(std::sync::PoisonError::into_inner);
209
210 let prior = overrides
211 .iter()
212 .map(|(key, _)| (key.to_string(), std::env::var(key).ok()))
213 .collect();
214
215 unsafe {
217 for (key, value) in overrides {
218 match value {
219 Some(value) => std::env::set_var(key, value),
220 None => std::env::remove_var(key),
221 }
222 }
223 }
224
225 Self {
226 _guard: guard,
227 prior,
228 }
229 }
230 }
231
232 impl Drop for ScopedEnv {
233 fn drop(&mut self) {
234 unsafe {
236 for (key, value) in self.prior.drain(..) {
237 match value {
238 Some(value) => std::env::set_var(&key, value),
239 None => std::env::remove_var(&key),
240 }
241 }
242 }
243 }
244 }
245
246 fn test_config() -> JobsConfig {
247 JobsConfig {
248 worker_id: "config-test-worker".to_string(),
249 poll_interval: Duration::from_millis(1),
250 claim_batch_size: 1,
251 lease_ttl_seconds: 1,
252 max_global_concurrency: 1,
253 reaper_interval: Duration::from_millis(1),
254 schedule_poll_interval: Duration::from_millis(1),
255 reaper_retry_delay_ms: 1,
256 }
257 }
258
259 #[test]
260 fn validate_accepts_minimum_direct_config_values() {
261 test_config()
262 .validate()
263 .expect("minimum direct config should be valid");
264 }
265
266 #[test]
267 fn validate_rejects_invalid_direct_config_values() {
268 let cases = [
269 {
270 let mut config = test_config();
271 config.worker_id = " ".to_string();
272 (config, JobsConfigValidationError::EmptyWorkerId)
273 },
274 {
275 let mut config = test_config();
276 config.poll_interval = Duration::ZERO;
277 (config, JobsConfigValidationError::ZeroPollInterval)
278 },
279 {
280 let mut config = test_config();
281 config.claim_batch_size = 0;
282 (
283 config,
284 JobsConfigValidationError::InvalidClaimBatchSize { actual: 0 },
285 )
286 },
287 {
288 let mut config = test_config();
289 config.lease_ttl_seconds = 0;
290 (
291 config,
292 JobsConfigValidationError::InvalidLeaseTtlSeconds { actual: 0 },
293 )
294 },
295 {
296 let mut config = test_config();
297 config.max_global_concurrency = 0;
298 (
299 config,
300 JobsConfigValidationError::InvalidMaxGlobalConcurrency,
301 )
302 },
303 {
304 let mut config = test_config();
305 config.reaper_interval = Duration::ZERO;
306 (config, JobsConfigValidationError::ZeroReaperInterval)
307 },
308 {
309 let mut config = test_config();
310 config.schedule_poll_interval = Duration::ZERO;
311 (config, JobsConfigValidationError::ZeroSchedulePollInterval)
312 },
313 {
314 let mut config = test_config();
315 config.reaper_retry_delay_ms = 0;
316 (
317 config,
318 JobsConfigValidationError::InvalidReaperRetryDelayMs { actual: 0 },
319 )
320 },
321 ];
322
323 for (config, expected) in cases {
324 assert_eq!(config.validate(), Err(expected));
325 }
326 }
327
328 #[test]
329 fn from_env_clamps_zero_intervals_to_non_zero_minimum() {
330 let _env = ScopedEnv::set(&[
331 ("JOBS_POLL_INTERVAL_MS", Some("0")),
332 ("JOBS_REAPER_INTERVAL_SECONDS", Some("0")),
333 ("JOBS_SCHEDULE_POLL_INTERVAL_SECONDS", Some("0")),
334 ]);
335
336 let config = JobsConfig::from_env();
337 assert_eq!(config.poll_interval, Duration::from_millis(1));
338 assert_eq!(config.reaper_interval, Duration::from_secs(1));
339 assert_eq!(config.schedule_poll_interval, Duration::from_secs(1));
340 }
341
342 #[test]
343 fn from_env_clamps_non_interval_limits_and_falls_back_worker_id() {
344 let _env = ScopedEnv::set(&[
345 ("JOBS_CLAIM_BATCH_SIZE", Some("0")),
346 ("JOBS_LEASE_TTL_SECONDS", Some("1")),
347 ("JOBS_MAX_GLOBAL_CONCURRENCY", Some("0")),
348 ("JOBS_REAPER_RETRY_DELAY_MS", Some("1")),
349 ("JOBS_WORKER_ID", Some(" ")),
350 ]);
351
352 let config = JobsConfig::from_env();
353 assert_eq!(config.claim_batch_size, 1);
354 assert_eq!(config.lease_ttl_seconds, 10);
355 assert_eq!(config.max_global_concurrency, 1);
356 assert_eq!(config.reaper_retry_delay_ms, 1_000);
357 assert!(config.worker_id.starts_with("worker-"));
358 }
359}