cronoq_core/
config.rs

1//! Engine configuration.
2
3use core::time::Duration;
4use std::sync::Arc;
5
6use crate::spec::{Priority, RetryPolicy};
7
8/// Per-class score parameters: `base + min(aging_rate * wait, age_cap) + urgency`.
9#[derive(Clone, Debug, PartialEq)]
10pub struct PriorityConfig {
11    /// Starting score per class.
12    pub base: [f64; Priority::COUNT],
13    /// Score points per second of waiting.
14    pub aging_rate: [f64; Priority::COUNT],
15    /// Maximum score points gained from waiting.
16    pub age_cap: [f64; Priority::COUNT],
17}
18
19impl Default for PriorityConfig {
20    fn default() -> Self {
21        PriorityConfig {
22            base: [10_000.0, 1_000.0, 100.0, 10.0],
23            aging_rate: [2.0; 4],
24            age_cap: [800.0; 4],
25        }
26    }
27}
28
29/// Deadline urgency curve (monotonically increasing as slack shrinks).
30#[derive(Clone, Debug, PartialEq)]
31pub struct UrgencyConfig {
32    /// Urgency reached when slack hits zero.
33    pub max: f64,
34    /// Minimum horizon before the latest start time where urgency starts rising.
35    pub window: Duration,
36    /// Extra urgency per second past the latest start time.
37    pub overdue_rate: f64,
38    /// Maximum extra urgency from being overdue.
39    pub overdue_cap: f64,
40}
41
42impl Default for UrgencyConfig {
43    fn default() -> Self {
44        UrgencyConfig {
45            max: 5_000.0,
46            window: Duration::from_secs(5),
47            overdue_rate: 100.0,
48            overdue_cap: 3_000.0,
49        }
50    }
51}
52
53/// Behaviour when a submission would exceed queue bounds.
54#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
55pub enum OverflowPolicy {
56    /// Reject with a retryable `Full` error; the runtime parks the caller until space frees up.
57    #[default]
58    Wait,
59    /// Reject with `Overloaded`.
60    Reject,
61    /// Evict the newest pending task of the lowest class strictly below the incoming one.
62    ShedLowest,
63    /// Evict the oldest pending `Low` task.
64    ShedOldest,
65}
66
67/// Maximum number of not-yet-running tasks.
68#[derive(Clone, Debug, Default, PartialEq, Eq)]
69pub struct Bounds {
70    /// Limit across all classes (`None` = unbounded).
71    pub max_pending: Option<usize>,
72    /// Limit per class (`None` = unbounded).
73    pub max_pending_per_class: [Option<usize>; Priority::COUNT],
74}
75
76/// How to treat a dependency on a key the engine has never seen.
77#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
78pub enum UnknownDependency {
79    /// Reject the submission with `UnknownDependency`.
80    #[default]
81    Reject,
82    /// Wait for a task with that key to be submitted; skip the dependent after `timeout`.
83    Await {
84        /// How long to wait for the key (`None` = forever).
85        timeout: Option<Duration>,
86    },
87}
88
89/// Queue-pressure autoscaling parameters (see SPEC §10).
90#[derive(Clone, Debug, PartialEq)]
91pub struct AutoscaleConfig {
92    /// Time between evaluations.
93    pub interval: Duration,
94    /// Utilization the arrival-rate term aims for (0..1].
95    pub target_utilization: f64,
96    /// Time in which the current backlog should be drained.
97    pub drain_time: Duration,
98    /// Maximum acceptable predicted queue wait per class (`None` = no SLO).
99    pub slo: [Option<Duration>; Priority::COUNT],
100    /// Reservations starting within this horizon count towards desired capacity.
101    pub reservation_horizon: Duration,
102    /// Scale-down uses the highest desired value seen within this window.
103    pub stabilization_window: Duration,
104    /// Minimum time between two scale-down steps.
105    pub cooldown: Duration,
106    /// Max relative growth per step (1.0 = +100%).
107    pub max_step_up_ratio: f64,
108    /// Minimum absolute growth allowed per step (the step is the larger of ratio and this).
109    pub max_step_up_abs: u32,
110    /// Max workers removed per scale-down step.
111    pub max_step_down: u32,
112    /// Smoothing factor for rate estimates (0..1, weight of the newest sample).
113    pub ewma_alpha: f64,
114    /// Cost of one worker-second, reported in snapshots.
115    pub cost_per_worker_second: f64,
116}
117
118impl Default for AutoscaleConfig {
119    fn default() -> Self {
120        AutoscaleConfig {
121            interval: Duration::from_secs(1),
122            target_utilization: 0.75,
123            drain_time: Duration::from_secs(30),
124            slo: [None; 4],
125            reservation_horizon: Duration::from_secs(60),
126            stabilization_window: Duration::from_secs(60),
127            cooldown: Duration::from_secs(30),
128            max_step_up_ratio: 1.0,
129            max_step_up_abs: 4,
130            max_step_down: 1,
131            ewma_alpha: 0.3,
132            cost_per_worker_second: 0.0,
133        }
134    }
135}
136
137/// A worker pool: capacity limits, scaling, retries, queue bounds and fixed-start behaviour.
138#[derive(Clone, Debug)]
139pub struct PoolConfig {
140    /// Unique pool name.
141    pub name: Arc<str>,
142    /// Lower bound on the autoscaler's desired capacity. Capacity itself may be lower, for
143    /// example at start (`initial_workers`) or when it is reported by remote workers.
144    pub min_workers: u32,
145    /// Upper capacity bound for the autoscaler (also sizes the reservation limit).
146    pub max_workers: u32,
147    /// Capacity at start (at most `max_workers`).
148    pub initial_workers: u32,
149    /// `None` disables autoscaling (capacity changes only via `CapacityChanged`).
150    pub autoscale: Option<AutoscaleConfig>,
151    /// Default retry policy for tasks without their own.
152    pub retry: RetryPolicy,
153    /// Pool queue bounds.
154    pub bounds: Bounds,
155    /// What to do when a bound is hit.
156    pub overflow: OverflowPolicy,
157    /// Fraction of `max_workers` that fixed-start reservations may occupy at once (at least 1 slot).
158    pub reservable_fraction: f64,
159    /// Duration estimate when nothing else is known.
160    pub default_duration: Duration,
161    /// Earliest a worker may be held free ahead of a fixed-start task.
162    pub max_hold: Duration,
163    /// Lease time-to-live; `Some` for remote workers that must heartbeat.
164    pub lease_ttl: Option<Duration>,
165    /// Handling of dependencies on unknown keys.
166    pub unknown_dependency: UnknownDependency,
167    /// Weighted fair sharing between tenants (`None`: tasks of a class run in submission
168    /// order whatever their tenant).
169    pub fairness: Option<Fairness>,
170}
171
172/// Weighted fair sharing of a pool between tenants ([`TaskSpec::tenant`](crate::TaskSpec::tenant)).
173///
174/// Within each priority class, tasks without a deadline are ordered so that each tenant with
175/// waiting work gets dispatches in proportion to its weight. Higher classes still run first,
176/// and tasks with deadlines keep their earliest-deadline order. A tenant that had no waiting
177/// work does not bank credit for later.
178#[derive(Clone, Debug, PartialEq)]
179pub struct Fairness {
180    /// Weights by tenant name.
181    pub weights: Vec<(Arc<str>, f64)>,
182    /// Weight of tenants not listed (and of tasks without a tenant).
183    pub default_weight: f64,
184}
185
186impl Default for Fairness {
187    fn default() -> Self {
188        Fairness {
189            weights: Vec::new(),
190            default_weight: 1.0,
191        }
192    }
193}
194
195impl Fairness {
196    /// Equal shares, with `weights` for the listed tenants.
197    pub fn weights<K: Into<Arc<str>>>(weights: impl IntoIterator<Item = (K, f64)>) -> Self {
198        Fairness {
199            weights: weights.into_iter().map(|(k, w)| (k.into(), w)).collect(),
200            ..Default::default()
201        }
202    }
203}
204
205impl PoolConfig {
206    /// Autoscaled pool with defaults: 1..=10 workers, starting at 1.
207    pub fn new(name: impl Into<Arc<str>>) -> Self {
208        PoolConfig {
209            name: name.into(),
210            min_workers: 1,
211            max_workers: 10,
212            initial_workers: 1,
213            autoscale: Some(AutoscaleConfig::default()),
214            retry: RetryPolicy::default(),
215            bounds: Bounds::default(),
216            overflow: OverflowPolicy::Wait,
217            reservable_fraction: 0.5,
218            default_duration: Duration::from_secs(1),
219            max_hold: Duration::from_secs(300),
220            lease_ttl: None,
221            unknown_dependency: UnknownDependency::Reject,
222            fairness: None,
223        }
224    }
225    /// Fixed worker count, no autoscaling.
226    pub fn fixed(name: impl Into<Arc<str>>, workers: u32) -> Self {
227        let mut p = Self::new(name);
228        p.min_workers = workers;
229        p.max_workers = workers;
230        p.initial_workers = workers;
231        p.autoscale = None;
232        p
233    }
234    /// Set min, initial and max worker counts.
235    pub fn workers(mut self, min: u32, initial: u32, max: u32) -> Self {
236        self.min_workers = min;
237        self.initial_workers = initial;
238        self.max_workers = max;
239        self
240    }
241    pub(crate) fn reservation_limit(&self) -> u32 {
242        ((self.max_workers as f64 * self.reservable_fraction).floor() as u32).max(1)
243    }
244}
245
246/// Kind of a shared resource.
247#[derive(Clone, Debug, PartialEq)]
248pub enum ResourceKind {
249    /// At most `permits` units held concurrently; released at the end of each attempt.
250    Semaphore {
251        /// Units available.
252        permits: u32,
253    },
254    /// Consumed units refill at `rate` per second up to `burst`.
255    TokenBucket {
256        /// Units added per second.
257        rate: f64,
258        /// Bucket size (also the initial fill).
259        burst: u32,
260    },
261}
262
263/// A named resource tasks can require (concurrency or rate limit), shared across pools.
264#[derive(Clone, Debug, PartialEq)]
265pub struct ResourceConfig {
266    /// Unique resource name.
267    pub name: Arc<str>,
268    /// Semaphore or token bucket.
269    pub kind: ResourceKind,
270}
271
272impl ResourceConfig {
273    /// A semaphore with `permits` units.
274    pub fn semaphore(name: impl Into<Arc<str>>, permits: u32) -> Self {
275        ResourceConfig {
276            name: name.into(),
277            kind: ResourceKind::Semaphore { permits },
278        }
279    }
280    /// A token bucket refilling `rate` units per second up to `burst`.
281    pub fn token_bucket(name: impl Into<Arc<str>>, rate: f64, burst: u32) -> Self {
282        ResourceConfig {
283            name: name.into(),
284            kind: ResourceKind::TokenBucket { rate, burst },
285        }
286    }
287}
288
289/// Complete engine configuration.
290#[derive(Clone, Debug)]
291pub struct CoreConfig {
292    /// Worker pools (at least one; the first is the default).
293    pub pools: Vec<PoolConfig>,
294    /// Shared resources.
295    pub resources: Vec<ResourceConfig>,
296    /// Class scores and aging.
297    pub priority: PriorityConfig,
298    /// Deadline urgency curve.
299    pub urgency: UrgencyConfig,
300    /// Queue bounds across all pools.
301    pub global_bounds: Bounds,
302    /// How long terminal statuses are remembered (for late dependents and lookups).
303    pub retention: Duration,
304    /// Maximum number of remembered terminal statuses.
305    pub retention_max: usize,
306    /// Emit `Action::Persist` mutations.
307    pub persist: bool,
308    /// Unix time (ns) corresponding to `Time::ZERO`; used for cron schedules.
309    pub epoch_unix_nanos: i64,
310    /// High bits mixed into task and lease ids (leader epoch in HA mode), so ids issued by
311    /// different leaders never collide.
312    pub id_epoch: u16,
313    /// Shard number, for servers that split pools between shards. When set, task ids carry it
314    /// in their top 8 bits (`shard << 56`, with the epoch and a sequence below it), so ids of
315    /// different shards never collide and name their shard ([`TaskId::shard`](crate::TaskId::shard)).
316    pub id_shard: Option<u8>,
317    /// Window for snapshot metrics.
318    pub metrics_window: Duration,
319    /// Seed for retry jitter.
320    pub seed: u64,
321}
322
323impl Default for CoreConfig {
324    fn default() -> Self {
325        CoreConfig {
326            pools: vec![PoolConfig::new("default")],
327            resources: Vec::new(),
328            priority: PriorityConfig::default(),
329            urgency: UrgencyConfig::default(),
330            global_bounds: Bounds::default(),
331            retention: Duration::from_secs(3600),
332            retention_max: 1_000_000,
333            persist: false,
334            epoch_unix_nanos: 0,
335            id_epoch: 0,
336            id_shard: None,
337            metrics_window: Duration::from_secs(60),
338            seed: 0x5eed_c0de,
339        }
340    }
341}
342
343impl CoreConfig {
344    /// Check the configuration for errors that [`Core::new`](crate::Core::new) would otherwise
345    /// reject with a panic (structural problems) or silently clamp (out-of-range numbers).
346    pub fn validate(&self) -> Result<(), String> {
347        if self.pools.is_empty() {
348            return Err("at least one pool is required".into());
349        }
350        let mut names = std::collections::HashSet::new();
351        for p in &self.pools {
352            if !names.insert(&p.name) {
353                return Err(format!("duplicate pool '{}'", p.name));
354            }
355            if p.min_workers > p.max_workers {
356                return Err(format!("pool '{}': min_workers > max_workers", p.name));
357            }
358            if !(p.reservable_fraction.is_finite() && p.reservable_fraction > 0.0) {
359                return Err(format!(
360                    "pool '{}': reservable_fraction must be > 0",
361                    p.name
362                ));
363            }
364            if p.retry.max_attempts == 0 {
365                return Err(format!(
366                    "pool '{}': retry.max_attempts must be >= 1",
367                    p.name
368                ));
369            }
370            if let Some(f) = &p.fairness {
371                let ok = |w: f64| w.is_finite() && w > 0.0;
372                if !ok(f.default_weight) || f.weights.iter().any(|(_, w)| !ok(*w)) {
373                    return Err(format!(
374                        "pool '{}': tenant weights must be finite and > 0",
375                        p.name
376                    ));
377                }
378            }
379            if let Some(a) = &p.autoscale {
380                if a.interval.is_zero() {
381                    return Err(format!("pool '{}': autoscale interval must be > 0", p.name));
382                }
383                if !(a.target_utilization > 0.0 && a.target_utilization <= 1.0) {
384                    return Err(format!(
385                        "pool '{}': target_utilization must be in (0, 1]",
386                        p.name
387                    ));
388                }
389                if !(a.ewma_alpha > 0.0 && a.ewma_alpha <= 1.0) {
390                    return Err(format!("pool '{}': ewma_alpha must be in (0, 1]", p.name));
391                }
392                if !(a.max_step_up_ratio.is_finite() && a.max_step_up_ratio >= 0.0) {
393                    return Err(format!("pool '{}': max_step_up_ratio must be >= 0", p.name));
394                }
395                if !(a.cost_per_worker_second.is_finite() && a.cost_per_worker_second >= 0.0) {
396                    return Err(format!(
397                        "pool '{}': cost_per_worker_second must be >= 0",
398                        p.name
399                    ));
400                }
401            }
402        }
403        let mut names = std::collections::HashSet::new();
404        for r in &self.resources {
405            if !names.insert(&r.name) {
406                return Err(format!("duplicate resource '{}'", r.name));
407            }
408            match r.kind {
409                ResourceKind::Semaphore { permits: 0 } => {
410                    return Err(format!("resource '{}': permits must be >= 1", r.name));
411                }
412                ResourceKind::TokenBucket { rate, burst } => {
413                    if rate.is_nan() || rate < 0.0 {
414                        return Err(format!("resource '{}': rate must be >= 0", r.name));
415                    }
416                    if burst == 0 {
417                        return Err(format!("resource '{}': burst must be >= 1", r.name));
418                    }
419                }
420                ResourceKind::Semaphore { .. } => {}
421            }
422        }
423        let finite = |v: &[f64]| v.iter().all(|x| x.is_finite());
424        if !finite(&self.priority.base)
425            || !finite(&self.priority.aging_rate)
426            || !finite(&self.priority.age_cap)
427        {
428            return Err("priority weights must be finite".into());
429        }
430        let u = &self.urgency;
431        if !finite(&[u.max, u.overdue_rate, u.overdue_cap]) {
432            return Err("urgency settings must be finite".into());
433        }
434        if self.metrics_window.is_zero() {
435            return Err("metrics_window must be > 0".into());
436        }
437        Ok(())
438    }
439
440    /// A copy with out-of-range numeric settings clamped to safe values, so the engine never
441    /// hangs (zero intervals) or panics (non-finite rates) on bad input.
442    pub(crate) fn sanitized(mut self) -> Self {
443        const MIN_INTERVAL: Duration = Duration::from_millis(1);
444        for p in &mut self.pools {
445            if p.reservable_fraction.is_nan() || p.reservable_fraction <= 0.0 {
446                p.reservable_fraction = 0.5;
447            }
448            if p.retry.max_attempts == 0 {
449                p.retry.max_attempts = 1;
450            }
451            if let Some(a) = &mut p.autoscale {
452                a.interval = a.interval.max(MIN_INTERVAL);
453                if !(a.target_utilization > 0.0 && a.target_utilization <= 1.0) {
454                    a.target_utilization = 0.75;
455                }
456                if !(a.ewma_alpha > 0.0 && a.ewma_alpha <= 1.0) {
457                    a.ewma_alpha = 0.3;
458                }
459                if !(a.max_step_up_ratio.is_finite() && a.max_step_up_ratio >= 0.0) {
460                    a.max_step_up_ratio = 1.0;
461                }
462                if !(a.cost_per_worker_second.is_finite() && a.cost_per_worker_second >= 0.0) {
463                    a.cost_per_worker_second = 0.0;
464                }
465            }
466        }
467        for r in &mut self.resources {
468            if let ResourceKind::TokenBucket { rate, .. } = &mut r.kind
469                && (rate.is_nan() || *rate < 0.0)
470            {
471                *rate = 0.0;
472            }
473        }
474        self.metrics_window = self.metrics_window.max(MIN_INTERVAL);
475        self
476    }
477}
478
479#[cfg(test)]
480mod tests {
481    use super::*;
482
483    #[test]
484    fn validate_reports_bad_settings() {
485        assert!(CoreConfig::default().validate().is_ok());
486        let mut c = CoreConfig::default();
487        c.pools[0].autoscale.as_mut().unwrap().interval = Duration::ZERO;
488        assert!(c.validate().unwrap_err().contains("interval"));
489        let c = CoreConfig {
490            resources: vec![ResourceConfig::token_bucket("x", f64::NAN, 1)],
491            ..Default::default()
492        };
493        assert!(c.validate().unwrap_err().contains("rate"));
494        let c = CoreConfig {
495            pools: vec![PoolConfig::new("a"), PoolConfig::new("a")],
496            ..Default::default()
497        };
498        assert!(c.validate().unwrap_err().contains("duplicate"));
499        let c = CoreConfig {
500            pools: vec![PoolConfig::new("a").workers(5, 1, 2)],
501            ..Default::default()
502        };
503        assert!(c.validate().is_err());
504    }
505}
cronoq_core/config.rs

cronoq_core/
config.rs
