runledger_runtime/
config.rs

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_INTENT_PROMOTER_POLL_INTERVAL_MS: u64 = 500;
9const DEFAULT_INTENT_PROMOTER_BATCH_SIZE: i64 = 16;
10const DEFAULT_LEASE_TTL_SECONDS: i32 = 60;
11const DEFAULT_MAX_GLOBAL_CONCURRENCY: usize = 32;
12const DEFAULT_REAPER_INTERVAL_SECONDS: u64 = 15;
13const DEFAULT_SCHEDULE_POLL_INTERVAL_SECONDS: u64 = 30;
14const DEFAULT_REAPER_RETRY_DELAY_MS: i32 = 30_000;
15
16/// Maximum batch size accepted by runtime worker, intent promoter, scheduler,
17/// and reaper loops.
18pub const JOBS_CLAIM_BATCH_SIZE_MAX: i64 = 1_000;
19
20#[derive(Debug, Clone)]
21pub struct JobsConfig {
22    pub worker_id: String,
23    pub poll_interval: Duration,
24    pub claim_batch_size: i64,
25    pub lease_ttl_seconds: i32,
26    pub max_global_concurrency: usize,
27    pub reaper_interval: Duration,
28    pub schedule_poll_interval: Duration,
29    pub reaper_retry_delay_ms: i32,
30}
31
32/// Polling and batch controls for durable enqueue-intent promotion.
33///
34/// [`crate::Supervisor`] derives these values from [`JobsConfig`] by default so
35/// existing deployments keep their current behavior. Use this type when intent
36/// traffic needs a cadence independent from ordinary queue claiming.
37#[derive(Debug, Clone, Copy, Eq, PartialEq)]
38pub struct IntentPromoterConfig {
39    poll_interval: Duration,
40    batch_size: i64,
41}
42
43impl IntentPromoterConfig {
44    #[must_use]
45    pub const fn new(poll_interval: Duration, batch_size: i64) -> Self {
46        Self {
47            poll_interval,
48            batch_size,
49        }
50    }
51
52    /// Reads intent-specific settings from the environment.
53    ///
54    /// `JOBS_INTENT_PROMOTER_POLL_INTERVAL_MS` defaults to 500 milliseconds and
55    /// `JOBS_INTENT_PROMOTER_BATCH_SIZE` defaults to 16.
56    #[must_use]
57    pub fn from_env() -> Self {
58        Self {
59            poll_interval: Duration::from_millis(
60                parse_env(
61                    "JOBS_INTENT_PROMOTER_POLL_INTERVAL_MS",
62                    DEFAULT_INTENT_PROMOTER_POLL_INTERVAL_MS,
63                )
64                .max(1),
65            ),
66            batch_size: parse_env(
67                "JOBS_INTENT_PROMOTER_BATCH_SIZE",
68                DEFAULT_INTENT_PROMOTER_BATCH_SIZE,
69            )
70            .clamp(1, JOBS_CLAIM_BATCH_SIZE_MAX),
71        }
72    }
73
74    /// Reads intent-specific settings from the environment, inheriting worker
75    /// settings for variables that are absent or invalid.
76    ///
77    /// This is the environment-composition policy used by
78    /// [`crate::Supervisor::builder_from_env`]. It preserves the default
79    /// coupling between worker claiming and intent promotion while allowing
80    /// either promoter control to be overridden independently.
81    #[must_use]
82    pub fn from_env_with_jobs_config_defaults(config: &JobsConfig) -> Self {
83        let poll_interval = parse_env_value::<u64>("JOBS_INTENT_PROMOTER_POLL_INTERVAL_MS")
84            .map(|milliseconds| Duration::from_millis(milliseconds.max(1)))
85            .unwrap_or(config.poll_interval);
86        let batch_size = parse_env_value::<i64>("JOBS_INTENT_PROMOTER_BATCH_SIZE")
87            .map(|batch_size| batch_size.clamp(1, JOBS_CLAIM_BATCH_SIZE_MAX))
88            .unwrap_or(config.claim_batch_size);
89
90        Self::new(poll_interval, batch_size)
91    }
92
93    #[must_use]
94    pub const fn from_jobs_config(config: &JobsConfig) -> Self {
95        Self::new(config.poll_interval, config.claim_batch_size)
96    }
97
98    pub fn validate(&self) -> Result<(), JobsConfigValidationError> {
99        if self.poll_interval.is_zero() {
100            return Err(JobsConfigValidationError::ZeroPollInterval);
101        }
102        validate_claim_batch_size(self.batch_size)
103    }
104
105    #[must_use]
106    pub const fn poll_interval(&self) -> Duration {
107        self.poll_interval
108    }
109
110    #[must_use]
111    pub const fn batch_size(&self) -> i64 {
112        self.batch_size
113    }
114}
115
116#[non_exhaustive]
117#[derive(Debug, Clone, Copy, Error, Eq, PartialEq)]
118pub enum JobsConfigValidationError {
119    #[error("jobs config worker_id must not be empty")]
120    EmptyWorkerId,
121    #[error("jobs config poll_interval must be greater than zero")]
122    ZeroPollInterval,
123    #[error("jobs config claim_batch_size must be between 1 and 1000, got {actual}")]
124    InvalidClaimBatchSize { actual: i64 },
125    #[error("jobs config lease_ttl_seconds must be at least 1, got {actual}")]
126    InvalidLeaseTtlSeconds { actual: i32 },
127    #[error("jobs config max_global_concurrency must be at least 1")]
128    InvalidMaxGlobalConcurrency,
129    #[error("jobs config reaper_interval must be greater than zero")]
130    ZeroReaperInterval,
131    #[error("jobs config schedule_poll_interval must be greater than zero")]
132    ZeroSchedulePollInterval,
133    #[error("jobs config reaper_retry_delay_ms must be at least 1, got {actual}")]
134    InvalidReaperRetryDelayMs { actual: i32 },
135}
136
137impl JobsConfig {
138    #[must_use]
139    pub fn from_env() -> Self {
140        Self {
141            worker_id: std::env::var("JOBS_WORKER_ID")
142                .ok()
143                .filter(|value| !value.trim().is_empty())
144                .unwrap_or_else(|| format!("worker-{}", uuid::Uuid::now_v7())),
145            poll_interval: Duration::from_millis(
146                parse_env("JOBS_POLL_INTERVAL_MS", DEFAULT_POLL_INTERVAL_MS).max(1),
147            ),
148            claim_batch_size: parse_env("JOBS_CLAIM_BATCH_SIZE", DEFAULT_CLAIM_BATCH_SIZE)
149                .clamp(1, JOBS_CLAIM_BATCH_SIZE_MAX),
150            lease_ttl_seconds: parse_env("JOBS_LEASE_TTL_SECONDS", DEFAULT_LEASE_TTL_SECONDS)
151                .max(10),
152            max_global_concurrency: parse_env(
153                "JOBS_MAX_GLOBAL_CONCURRENCY",
154                DEFAULT_MAX_GLOBAL_CONCURRENCY,
155            )
156            .max(1),
157            reaper_interval: Duration::from_secs(
158                parse_env(
159                    "JOBS_REAPER_INTERVAL_SECONDS",
160                    DEFAULT_REAPER_INTERVAL_SECONDS,
161                )
162                .max(1),
163            ),
164            schedule_poll_interval: Duration::from_secs(
165                parse_env(
166                    "JOBS_SCHEDULE_POLL_INTERVAL_SECONDS",
167                    DEFAULT_SCHEDULE_POLL_INTERVAL_SECONDS,
168                )
169                .max(1),
170            ),
171            reaper_retry_delay_ms: parse_env(
172                "JOBS_REAPER_RETRY_DELAY_MS",
173                DEFAULT_REAPER_RETRY_DELAY_MS,
174            )
175            .max(1_000),
176        }
177    }
178
179    pub fn validate(&self) -> Result<(), JobsConfigValidationError> {
180        if self.worker_id.trim().is_empty() {
181            return Err(JobsConfigValidationError::EmptyWorkerId);
182        }
183        if self.poll_interval.is_zero() {
184            return Err(JobsConfigValidationError::ZeroPollInterval);
185        }
186        validate_claim_batch_size(self.claim_batch_size)?;
187        if self.lease_ttl_seconds < 1 {
188            return Err(JobsConfigValidationError::InvalidLeaseTtlSeconds {
189                actual: self.lease_ttl_seconds,
190            });
191        }
192        if self.max_global_concurrency < 1 {
193            return Err(JobsConfigValidationError::InvalidMaxGlobalConcurrency);
194        }
195        if self.reaper_interval.is_zero() {
196            return Err(JobsConfigValidationError::ZeroReaperInterval);
197        }
198        if self.schedule_poll_interval.is_zero() {
199            return Err(JobsConfigValidationError::ZeroSchedulePollInterval);
200        }
201        if self.reaper_retry_delay_ms < 1 {
202            return Err(JobsConfigValidationError::InvalidReaperRetryDelayMs {
203                actual: self.reaper_retry_delay_ms,
204            });
205        }
206
207        Ok(())
208    }
209
210    pub(crate) fn validate_worker_loop(&self) -> Result<(), JobsConfigValidationError> {
211        if self.worker_id.trim().is_empty() {
212            return Err(JobsConfigValidationError::EmptyWorkerId);
213        }
214        if self.poll_interval.is_zero() {
215            return Err(JobsConfigValidationError::ZeroPollInterval);
216        }
217        validate_claim_batch_size(self.claim_batch_size)?;
218        if self.lease_ttl_seconds < 1 {
219            return Err(JobsConfigValidationError::InvalidLeaseTtlSeconds {
220                actual: self.lease_ttl_seconds,
221            });
222        }
223        if self.max_global_concurrency < 1 {
224            return Err(JobsConfigValidationError::InvalidMaxGlobalConcurrency);
225        }
226
227        Ok(())
228    }
229
230    pub(crate) fn validate_scheduler_loop(&self) -> Result<(), JobsConfigValidationError> {
231        validate_claim_batch_size(self.claim_batch_size)?;
232        if self.schedule_poll_interval.is_zero() {
233            return Err(JobsConfigValidationError::ZeroSchedulePollInterval);
234        }
235
236        Ok(())
237    }
238
239    pub(crate) fn validate_reaper_loop(&self) -> Result<(), JobsConfigValidationError> {
240        validate_claim_batch_size(self.claim_batch_size)?;
241        if self.reaper_interval.is_zero() {
242            return Err(JobsConfigValidationError::ZeroReaperInterval);
243        }
244        if self.reaper_retry_delay_ms < 1 {
245            return Err(JobsConfigValidationError::InvalidReaperRetryDelayMs {
246                actual: self.reaper_retry_delay_ms,
247            });
248        }
249
250        Ok(())
251    }
252}
253
254fn validate_claim_batch_size(claim_batch_size: i64) -> Result<(), JobsConfigValidationError> {
255    if (1..=JOBS_CLAIM_BATCH_SIZE_MAX).contains(&claim_batch_size) {
256        return Ok(());
257    }
258
259    Err(JobsConfigValidationError::InvalidClaimBatchSize {
260        actual: claim_batch_size,
261    })
262}
263
264fn parse_env<T>(name: &str, default: T) -> T
265where
266    T: FromStr,
267{
268    parse_env_value(name).unwrap_or(default)
269}
270
271fn parse_env_value<T>(name: &str) -> Option<T>
272where
273    T: FromStr,
274{
275    std::env::var(name)
276        .ok()
277        .and_then(|value| value.parse::<T>().ok())
278}
279
280#[cfg(test)]
281mod tests {
282    use std::sync::{Mutex, OnceLock};
283
284    use super::*;
285
286    static ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
287
288    #[derive(Debug)]
289    struct ScopedEnv {
290        _guard: std::sync::MutexGuard<'static, ()>,
291        prior: Vec<(String, Option<String>)>,
292    }
293
294    impl ScopedEnv {
295        #[allow(
296            unsafe_code,
297            reason = "Rust 2024 requires unsafe environment mutation, serialized here by ENV_LOCK"
298        )]
299        fn set(overrides: &[(&str, Option<&str>)]) -> Self {
300            let guard = ENV_LOCK
301                .get_or_init(|| Mutex::new(()))
302                .lock()
303                .unwrap_or_else(std::sync::PoisonError::into_inner);
304
305            let prior = overrides
306                .iter()
307                .map(|(key, _)| (key.to_string(), std::env::var(key).ok()))
308                .collect();
309
310            // SAFETY: env mutation is serialized through ENV_LOCK.
311            unsafe {
312                for (key, value) in overrides {
313                    match value {
314                        Some(value) => std::env::set_var(key, value),
315                        None => std::env::remove_var(key),
316                    }
317                }
318            }
319
320            Self {
321                _guard: guard,
322                prior,
323            }
324        }
325    }
326
327    impl Drop for ScopedEnv {
328        #[allow(
329            unsafe_code,
330            reason = "Rust 2024 requires unsafe environment mutation, serialized here by the held ENV_LOCK guard"
331        )]
332        fn drop(&mut self) {
333            // SAFETY: env mutation is serialized through ENV_LOCK.
334            unsafe {
335                for (key, value) in self.prior.drain(..) {
336                    match value {
337                        Some(value) => std::env::set_var(&key, value),
338                        None => std::env::remove_var(&key),
339                    }
340                }
341            }
342        }
343    }
344
345    fn test_config() -> JobsConfig {
346        JobsConfig {
347            worker_id: "config-test-worker".to_string(),
348            poll_interval: Duration::from_millis(1),
349            claim_batch_size: 1,
350            lease_ttl_seconds: 1,
351            max_global_concurrency: 1,
352            reaper_interval: Duration::from_millis(1),
353            schedule_poll_interval: Duration::from_millis(1),
354            reaper_retry_delay_ms: 1,
355        }
356    }
357
358    #[test]
359    fn validate_accepts_minimum_direct_config_values() {
360        test_config()
361            .validate()
362            .expect("minimum direct config should be valid");
363    }
364
365    #[test]
366    fn intent_promoter_config_validates_independently() {
367        let jobs_config = test_config();
368        assert_eq!(
369            IntentPromoterConfig::from_jobs_config(&jobs_config),
370            IntentPromoterConfig::new(jobs_config.poll_interval, jobs_config.claim_batch_size)
371        );
372        assert_eq!(
373            IntentPromoterConfig::new(Duration::ZERO, 1).validate(),
374            Err(JobsConfigValidationError::ZeroPollInterval)
375        );
376        assert_eq!(
377            IntentPromoterConfig::new(Duration::from_millis(1), 0).validate(),
378            Err(JobsConfigValidationError::InvalidClaimBatchSize { actual: 0 })
379        );
380    }
381
382    #[test]
383    fn validate_rejects_invalid_direct_config_values() {
384        let cases = [
385            {
386                let mut config = test_config();
387                config.worker_id = "  ".to_string();
388                (config, JobsConfigValidationError::EmptyWorkerId)
389            },
390            {
391                let mut config = test_config();
392                config.poll_interval = Duration::ZERO;
393                (config, JobsConfigValidationError::ZeroPollInterval)
394            },
395            {
396                let mut config = test_config();
397                config.claim_batch_size = 0;
398                (
399                    config,
400                    JobsConfigValidationError::InvalidClaimBatchSize { actual: 0 },
401                )
402            },
403            {
404                let mut config = test_config();
405                config.claim_batch_size = JOBS_CLAIM_BATCH_SIZE_MAX + 1;
406                (
407                    config,
408                    JobsConfigValidationError::InvalidClaimBatchSize {
409                        actual: JOBS_CLAIM_BATCH_SIZE_MAX + 1,
410                    },
411                )
412            },
413            {
414                let mut config = test_config();
415                config.lease_ttl_seconds = 0;
416                (
417                    config,
418                    JobsConfigValidationError::InvalidLeaseTtlSeconds { actual: 0 },
419                )
420            },
421            {
422                let mut config = test_config();
423                config.max_global_concurrency = 0;
424                (
425                    config,
426                    JobsConfigValidationError::InvalidMaxGlobalConcurrency,
427                )
428            },
429            {
430                let mut config = test_config();
431                config.reaper_interval = Duration::ZERO;
432                (config, JobsConfigValidationError::ZeroReaperInterval)
433            },
434            {
435                let mut config = test_config();
436                config.schedule_poll_interval = Duration::ZERO;
437                (config, JobsConfigValidationError::ZeroSchedulePollInterval)
438            },
439            {
440                let mut config = test_config();
441                config.reaper_retry_delay_ms = 0;
442                (
443                    config,
444                    JobsConfigValidationError::InvalidReaperRetryDelayMs { actual: 0 },
445                )
446            },
447        ];
448
449        for (config, expected) in cases {
450            assert_eq!(config.validate(), Err(expected));
451        }
452    }
453
454    #[test]
455    fn from_env_clamps_zero_intervals_to_non_zero_minimum() {
456        let _env = ScopedEnv::set(&[
457            ("JOBS_POLL_INTERVAL_MS", Some("0")),
458            ("JOBS_REAPER_INTERVAL_SECONDS", Some("0")),
459            ("JOBS_SCHEDULE_POLL_INTERVAL_SECONDS", Some("0")),
460        ]);
461
462        let config = JobsConfig::from_env();
463        assert_eq!(config.poll_interval, Duration::from_millis(1));
464        assert_eq!(config.reaper_interval, Duration::from_secs(1));
465        assert_eq!(config.schedule_poll_interval, Duration::from_secs(1));
466    }
467
468    #[test]
469    fn intent_promoter_from_env_uses_independent_controls() {
470        let _env = ScopedEnv::set(&[
471            ("JOBS_INTENT_PROMOTER_POLL_INTERVAL_MS", Some("37")),
472            ("JOBS_INTENT_PROMOTER_BATCH_SIZE", Some("9")),
473        ]);
474
475        let config = IntentPromoterConfig::from_env();
476        assert_eq!(config.poll_interval(), Duration::from_millis(37));
477        assert_eq!(config.batch_size(), 9);
478    }
479
480    #[test]
481    fn intent_promoter_env_overrides_fall_back_to_jobs_config_independently() {
482        let _env = ScopedEnv::set(&[
483            ("JOBS_INTENT_PROMOTER_POLL_INTERVAL_MS", Some("37")),
484            ("JOBS_INTENT_PROMOTER_BATCH_SIZE", None),
485        ]);
486        let mut jobs_config = test_config();
487        jobs_config.poll_interval = Duration::from_millis(83);
488        jobs_config.claim_batch_size = 7;
489
490        let config = IntentPromoterConfig::from_env_with_jobs_config_defaults(&jobs_config);
491        assert_eq!(config.poll_interval(), Duration::from_millis(37));
492        assert_eq!(config.batch_size(), 7);
493    }
494
495    #[test]
496    fn from_env_clamps_non_interval_limits_and_falls_back_worker_id() {
497        let _env = ScopedEnv::set(&[
498            ("JOBS_CLAIM_BATCH_SIZE", Some("1001")),
499            ("JOBS_LEASE_TTL_SECONDS", Some("1")),
500            ("JOBS_MAX_GLOBAL_CONCURRENCY", Some("0")),
501            ("JOBS_REAPER_RETRY_DELAY_MS", Some("1")),
502            ("JOBS_WORKER_ID", Some("   ")),
503        ]);
504
505        let config = JobsConfig::from_env();
506        assert_eq!(config.claim_batch_size, JOBS_CLAIM_BATCH_SIZE_MAX);
507        assert_eq!(config.lease_ttl_seconds, 10);
508        assert_eq!(config.max_global_concurrency, 1);
509        assert_eq!(config.reaper_retry_delay_ms, 1_000);
510        assert!(config.worker_id.starts_with("worker-"));
511    }
512}
runledger_runtime/config.rs

runledger_runtime/
config.rs
