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