use serde::{Deserialize, Serialize};
use std::time::Duration;
fn default_concurrency() -> usize {
std::thread::available_parallelism()
.map(|n| n.get() * 2)
.unwrap_or(8)
.clamp(4, 16)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchedulerConfig {
pub max_concurrency: usize,
pub quota_per_run: u32,
pub retry: RetryPolicy,
}
impl Default for SchedulerConfig {
fn default() -> Self {
Self {
max_concurrency: default_concurrency(),
quota_per_run: 1000,
retry: RetryPolicy::default(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryPolicy {
pub max_attempts: u32,
pub initial_backoff: Duration,
pub backoff_multiplier: f64,
pub max_backoff: Duration,
pub schema_retry_max: u32,
}
impl Default for RetryPolicy {
fn default() -> Self {
Self {
max_attempts: 2,
initial_backoff: Duration::from_millis(500),
backoff_multiplier: 2.0,
max_backoff: Duration::from_secs(10),
schema_retry_max: 3,
}
}
}
impl RetryPolicy {
pub fn backoff(&self, attempt: u32) -> Duration {
let exp = attempt.saturating_sub(1) as i32;
let secs = self.initial_backoff.as_secs_f64() * self.backoff_multiplier.powi(exp);
Duration::from_secs_f64(secs.min(self.max_backoff.as_secs_f64()))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn scheduler_config_default_is_in_adaptive_range() {
let cfg = SchedulerConfig::default();
assert!(
(4..=16).contains(&cfg.max_concurrency),
"default max_concurrency {} outside adaptive [4,16]",
cfg.max_concurrency
);
assert_eq!(cfg.quota_per_run, 1000);
assert_eq!(cfg.retry.max_attempts, 2);
assert_eq!(cfg.retry.initial_backoff, Duration::from_millis(500));
assert_eq!(cfg.retry.schema_retry_max, 3);
}
#[test]
fn scheduler_config_clone_preserves_fields() {
let cfg = SchedulerConfig {
max_concurrency: 32,
quota_per_run: 250,
retry: RetryPolicy {
max_attempts: 5,
initial_backoff: Duration::from_millis(100),
backoff_multiplier: 1.5,
max_backoff: Duration::from_secs(20),
schema_retry_max: 1,
},
};
let cloned = cfg.clone();
assert_eq!(cloned.max_concurrency, 32);
assert_eq!(cloned.quota_per_run, 250);
assert_eq!(cloned.retry.max_attempts, 5);
assert_eq!(cloned.retry.initial_backoff, Duration::from_millis(100));
assert_eq!(cloned.retry.schema_retry_max, 1);
}
#[test]
fn scheduler_config_serde_roundtrip() {
let cfg = SchedulerConfig {
max_concurrency: 8,
quota_per_run: 100,
retry: RetryPolicy {
max_attempts: 3,
initial_backoff: Duration::from_millis(250),
backoff_multiplier: 1.7,
max_backoff: Duration::from_secs(5),
schema_retry_max: 2,
},
};
let json = serde_json::to_string(&cfg).unwrap();
let back: SchedulerConfig = serde_json::from_str(&json).unwrap();
assert_eq!(back.max_concurrency, cfg.max_concurrency);
assert_eq!(back.quota_per_run, cfg.quota_per_run);
assert_eq!(back.retry.max_attempts, cfg.retry.max_attempts);
assert_eq!(back.retry.initial_backoff, cfg.retry.initial_backoff);
assert_eq!(back.retry.backoff_multiplier, cfg.retry.backoff_multiplier);
assert_eq!(back.retry.max_backoff, cfg.retry.max_backoff);
assert_eq!(back.retry.schema_retry_max, cfg.retry.schema_retry_max);
}
#[test]
fn scheduler_config_debug_format_includes_field_names() {
let cfg = SchedulerConfig::default();
let dbg = format!("{:?}", cfg);
assert!(dbg.contains("max_concurrency"));
assert!(dbg.contains("quota_per_run"));
assert!(dbg.contains("retry"));
}
#[test]
fn retry_policy_default_values() {
let r = RetryPolicy::default();
assert_eq!(r.max_attempts, 2);
assert_eq!(r.initial_backoff, Duration::from_millis(500));
assert_eq!(r.backoff_multiplier, 2.0);
assert_eq!(r.max_backoff, Duration::from_secs(10));
assert_eq!(r.schema_retry_max, 3);
}
#[test]
fn retry_policy_serde_roundtrip() {
let r = RetryPolicy {
max_attempts: 4,
initial_backoff: Duration::from_millis(123),
backoff_multiplier: 3.5,
max_backoff: Duration::from_secs(30),
schema_retry_max: 0,
};
let json = serde_json::to_string(&r).unwrap();
let back: RetryPolicy = serde_json::from_str(&json).unwrap();
assert_eq!(back.max_attempts, r.max_attempts);
assert_eq!(back.initial_backoff, r.initial_backoff);
assert_eq!(back.backoff_multiplier, r.backoff_multiplier);
assert_eq!(back.max_backoff, r.max_backoff);
assert_eq!(back.schema_retry_max, r.schema_retry_max);
}
#[test]
fn backoff_attempt_1_returns_initial_backoff() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(500),
backoff_multiplier: 2.0,
max_backoff: Duration::from_secs(60),
..RetryPolicy::default()
};
assert_eq!(r.backoff(1), Duration::from_millis(500));
}
#[test]
fn backoff_doubles_each_attempt() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(100),
backoff_multiplier: 2.0,
max_backoff: Duration::from_secs(60),
..RetryPolicy::default()
};
assert_eq!(r.backoff(1), Duration::from_millis(100));
assert_eq!(r.backoff(2), Duration::from_millis(200));
assert_eq!(r.backoff(3), Duration::from_millis(400));
assert_eq!(r.backoff(4), Duration::from_millis(800));
assert_eq!(r.backoff(5), Duration::from_millis(1600));
}
#[test]
fn backoff_capped_at_max_backoff() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(500),
backoff_multiplier: 2.0,
max_backoff: Duration::from_secs(1),
..RetryPolicy::default()
};
assert_eq!(r.backoff(1), Duration::from_millis(500));
assert_eq!(r.backoff(2), Duration::from_secs(1));
assert_eq!(r.backoff(10), Duration::from_secs(1));
assert_eq!(r.backoff(60), Duration::from_secs(1));
}
#[test]
fn backoff_attempt_0_treated_as_attempt_1() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(750),
backoff_multiplier: 2.0,
max_backoff: Duration::from_secs(60),
..RetryPolicy::default()
};
assert_eq!(r.backoff(0), r.backoff(1));
assert_eq!(r.backoff(0), Duration::from_millis(750));
}
#[test]
fn backoff_fractional_multiplier() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(1000),
backoff_multiplier: 1.5,
max_backoff: Duration::from_secs(60),
..RetryPolicy::default()
};
let b1 = r.backoff(1);
let b2 = r.backoff(2);
let b3 = r.backoff(3);
assert_eq!(b1, Duration::from_millis(1000));
assert_eq!(b2, Duration::from_millis(1500));
assert_eq!(b3, Duration::from_millis(2250));
assert!(b1 < b2);
assert!(b2 < b3);
}
#[test]
fn backoff_never_exceeds_max_backoff_even_with_large_multiplier() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(10),
backoff_multiplier: 100.0,
max_backoff: Duration::from_secs(2),
..RetryPolicy::default()
};
for attempt in 1..20 {
let b = r.backoff(attempt);
assert!(
b <= r.max_backoff,
"backoff for attempt {} ({:?}) exceeded cap {:?}",
attempt,
b,
r.max_backoff
);
}
}
#[test]
fn backoff_zero_initial_returns_zero() {
let r = RetryPolicy {
initial_backoff: Duration::ZERO,
backoff_multiplier: 2.0,
max_backoff: Duration::from_secs(60),
..RetryPolicy::default()
};
assert_eq!(r.backoff(1), Duration::ZERO);
assert_eq!(r.backoff(5), Duration::ZERO);
}
#[test]
fn backoff_multiplier_one_keeps_constant() {
let r = RetryPolicy {
initial_backoff: Duration::from_millis(250),
backoff_multiplier: 1.0,
max_backoff: Duration::from_secs(60),
..RetryPolicy::default()
};
let b1 = r.backoff(1);
for attempt in 2..=10 {
assert_eq!(r.backoff(attempt), b1);
}
}
#[test]
fn retry_policy_clone_preserves_fields() {
let r = RetryPolicy {
max_attempts: 7,
initial_backoff: Duration::from_millis(123),
backoff_multiplier: 1.25,
max_backoff: Duration::from_secs(45),
schema_retry_max: 0,
};
let cloned = r.clone();
assert_eq!(cloned.max_attempts, r.max_attempts);
assert_eq!(cloned.initial_backoff, r.initial_backoff);
assert_eq!(cloned.backoff_multiplier, r.backoff_multiplier);
assert_eq!(cloned.max_backoff, r.max_backoff);
assert_eq!(cloned.schema_retry_max, r.schema_retry_max);
}
#[test]
fn retry_policy_debug_format() {
let r = RetryPolicy::default();
let dbg = format!("{:?}", r);
assert!(dbg.contains("max_attempts"));
assert!(dbg.contains("initial_backoff"));
assert!(dbg.contains("backoff_multiplier"));
assert!(dbg.contains("max_backoff"));
assert!(dbg.contains("schema_retry_max"));
}
}