use std::num::NonZeroU32;
use std::time::Duration;
use crate::{
core::{ConfigError, TaskDefaults},
policies::BackoffPolicy,
policies::RestartPolicy,
tasks::task::TaskRef,
};
#[inline]
fn normalize_timeout(timeout: Option<Duration>) -> Option<Duration> {
timeout.filter(|d| !d.is_zero())
}
#[derive(Clone)]
#[must_use]
pub struct TaskSpec {
restart: TaskSetting<RestartPolicy>,
backoff: TaskSetting<BackoffPolicy>,
timeout: TaskSetting<Option<Duration>>,
max_retries: TaskSetting<Option<NonZeroU32>>,
task: TaskRef,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TaskSetting<T> {
Inherit,
Explicit(T),
}
impl<T: Copy> TaskSetting<T> {
#[inline]
fn value(self) -> Option<T> {
match self {
Self::Inherit => None,
Self::Explicit(value) => Some(value),
}
}
#[inline]
fn resolve(self, default: T) -> T {
match self {
Self::Inherit => default,
Self::Explicit(value) => value,
}
}
}
#[derive(Clone)]
#[must_use]
pub(crate) struct ResolvedTaskSpec {
restart: RestartPolicy,
backoff: BackoffPolicy,
timeout: Option<Duration>,
max_retries: Option<NonZeroU32>,
task: TaskRef,
}
impl std::fmt::Debug for TaskSpec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("TaskSpec")
.field("restart", &self.restart)
.field("backoff", &self.backoff)
.field("timeout", &self.timeout)
.field("task", &self.task.name())
.field("max_retries", &self.max_retries)
.finish()
}
}
impl std::fmt::Debug for ResolvedTaskSpec {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ResolvedTaskSpec")
.field("restart", &self.restart)
.field("backoff", &self.backoff)
.field("timeout", &self.timeout)
.field("task", &self.task.name())
.field("max_retries", &self.max_retries)
.finish()
}
}
impl TaskSpec {
pub fn from_defaults(task: TaskRef) -> Self {
Self {
restart: TaskSetting::Inherit,
backoff: TaskSetting::Inherit,
timeout: TaskSetting::Inherit,
max_retries: TaskSetting::Inherit,
task,
}
}
pub fn new(
task: TaskRef,
restart: RestartPolicy,
backoff: BackoffPolicy,
timeout: impl Into<Option<Duration>>,
) -> Self {
Self {
restart: TaskSetting::Explicit(restart),
backoff: TaskSetting::Explicit(backoff),
timeout: TaskSetting::Explicit(normalize_timeout(timeout.into())),
max_retries: TaskSetting::Explicit(None),
task,
}
}
pub fn once(task: TaskRef) -> Self {
Self {
restart: TaskSetting::Explicit(RestartPolicy::Never),
backoff: TaskSetting::Inherit,
timeout: TaskSetting::Inherit,
max_retries: TaskSetting::Inherit,
task,
}
}
pub fn restartable(task: TaskRef) -> Self {
Self {
restart: TaskSetting::Explicit(RestartPolicy::OnFailure),
backoff: TaskSetting::Inherit,
timeout: TaskSetting::Inherit,
max_retries: TaskSetting::Inherit,
task,
}
}
#[doc(alias = "interval")]
#[doc(alias = "fixed delay")]
pub fn periodic(task: TaskRef, every: Duration) -> Self {
Self {
restart: TaskSetting::Explicit(RestartPolicy::Always {
interval: Some(every).filter(|d| !d.is_zero()),
}),
backoff: TaskSetting::Inherit,
timeout: TaskSetting::Inherit,
max_retries: TaskSetting::Inherit,
task,
}
}
#[must_use]
pub fn task(&self) -> &TaskRef {
&self.task
}
#[must_use]
pub fn name(&self) -> &str {
self.task.name()
}
#[must_use]
pub fn restart_override(&self) -> Option<RestartPolicy> {
self.restart.value()
}
#[must_use]
pub fn backoff_override(&self) -> Option<BackoffPolicy> {
self.backoff.value()
}
#[must_use]
pub fn timeout_override(&self) -> TaskSetting<Option<Duration>> {
self.timeout
}
#[must_use]
pub fn max_retries_override(&self) -> TaskSetting<Option<NonZeroU32>> {
self.max_retries
}
#[doc(alias = "watchdog")]
#[doc(alias = "attempt deadline")]
pub fn with_timeout(mut self, timeout: impl Into<Option<Duration>>) -> Self {
self.timeout = TaskSetting::Explicit(normalize_timeout(timeout.into()));
self
}
pub fn with_backoff(mut self, backoff: BackoffPolicy) -> Self {
self.backoff = TaskSetting::Explicit(backoff);
self
}
pub fn with_restart(mut self, restart: RestartPolicy) -> Self {
self.restart = TaskSetting::Explicit(restart);
self
}
#[doc(alias = "retry limit")]
#[doc(alias = "retry budget")]
pub fn with_max_retries(mut self, max_retries: impl Into<Option<NonZeroU32>>) -> Self {
self.max_retries = TaskSetting::Explicit(max_retries.into());
self
}
pub fn try_with_max_retries(self, max_retries: u32) -> Result<Self, ConfigError> {
let max_retries = NonZeroU32::new(max_retries).ok_or(ConfigError::Zero {
field: "max_retries",
})?;
Ok(self.with_max_retries(max_retries))
}
pub(crate) fn resolve(self, defaults: &TaskDefaults) -> ResolvedTaskSpec {
ResolvedTaskSpec {
restart: self.restart.resolve(defaults.restart()),
backoff: self.backoff.resolve(defaults.backoff()),
timeout: self.timeout.resolve(defaults.timeout()),
max_retries: self.max_retries.resolve(defaults.max_retries()),
task: self.task,
}
}
}
impl ResolvedTaskSpec {
pub(crate) fn task(&self) -> &TaskRef {
&self.task
}
#[cfg(test)]
pub(crate) fn name(&self) -> &str {
self.task.name()
}
pub(crate) fn restart(&self) -> RestartPolicy {
self.restart
}
pub(crate) fn backoff(&self) -> BackoffPolicy {
self.backoff
}
pub(crate) fn timeout(&self) -> Option<Duration> {
self.timeout
}
pub(crate) fn max_retries(&self) -> Option<NonZeroU32> {
self.max_retries
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{JitterPolicy, TaskContext, TaskFn};
fn task(name: &str) -> TaskRef {
TaskFn::arc(name, |_ctx: TaskContext| async { Ok(()) })
}
fn assert_inherits_non_restart_settings(spec: &TaskSpec) {
assert!(spec.backoff_override().is_none());
assert_eq!(spec.timeout_override(), TaskSetting::Inherit);
assert_eq!(spec.max_retries_override(), TaskSetting::Inherit);
}
fn assert_explicit_timeout(spec: TaskSpec, expected: Option<Duration>, case: &str) {
assert_eq!(
spec.timeout_override(),
TaskSetting::Explicit(expected),
"{case}"
);
}
#[test]
fn named_constructors_set_restart_and_inherit_other_settings() {
let inherited = TaskSpec::from_defaults(task("inherited"));
assert!(inherited.restart_override().is_none());
assert_inherits_non_restart_settings(&inherited);
let once = TaskSpec::once(task("once"));
assert!(matches!(
once.restart_override(),
Some(RestartPolicy::Never)
));
assert_inherits_non_restart_settings(&once);
let restartable = TaskSpec::restartable(task("restartable"));
assert!(matches!(
restartable.restart_override(),
Some(RestartPolicy::OnFailure)
));
assert_inherits_non_restart_settings(&restartable);
let every = Duration::from_secs(30);
let spec = TaskSpec::periodic(task("tick"), every);
assert!(
matches!(spec.restart_override(), Some(RestartPolicy::Always { interval: Some(d) }) if d == every),
"periodic must set RestartPolicy::Always with the given interval, got {:?}",
spec.restart_override()
);
assert_inherits_non_restart_settings(&spec);
let immediate = TaskSpec::periodic(task("immediate"), Duration::ZERO);
assert!(
matches!(
immediate.restart_override(),
Some(RestartPolicy::Always { interval: None })
),
"a zero interval must normalize to None (immediate restart), got {:?}",
immediate.restart_override()
);
}
#[test]
fn new_marks_every_setting_as_explicit() {
let backoff = BackoffPolicy::constant(Duration::from_secs(2));
let timeout = Duration::from_secs(7);
let spec = TaskSpec::new(task("explicit"), RestartPolicy::Never, backoff, timeout);
assert!(matches!(
spec.restart_override(),
Some(RestartPolicy::Never)
));
assert_eq!(
spec.backoff_override().map(|policy| policy.first()),
Some(Duration::from_secs(2))
);
assert_eq!(
spec.timeout_override(),
TaskSetting::Explicit(Some(timeout))
);
assert_eq!(spec.max_retries_override(), TaskSetting::Explicit(None));
}
#[test]
fn explicit_none_disables_inherited_optional_settings() {
let retries = NonZeroU32::new(4).unwrap();
let defaults = TaskDefaults::default()
.with_timeout(Duration::from_secs(9))
.with_max_retries(retries);
let spec = TaskSpec::restartable(task("disabled"))
.with_timeout(None)
.with_max_retries(None);
assert_eq!(spec.timeout_override(), TaskSetting::Explicit(None));
assert_eq!(spec.max_retries_override(), TaskSetting::Explicit(None));
let resolved = spec.resolve(&defaults);
assert_eq!(resolved.timeout(), None);
assert_eq!(resolved.max_retries(), None);
}
#[test]
fn resolve_applies_defaults_only_to_inherited_settings() {
let retries = NonZeroU32::new(6).unwrap();
let defaults = TaskDefaults::default()
.with_restart(RestartPolicy::Never)
.with_backoff(BackoffPolicy::constant(Duration::from_secs(3)))
.with_timeout(Duration::from_secs(12))
.with_max_retries(retries);
let spec = TaskSpec::restartable(task("worker"));
let resolved = spec.resolve(&defaults);
assert_eq!(resolved.name(), "worker");
assert_eq!(resolved.task().name(), "worker");
assert!(matches!(resolved.restart(), RestartPolicy::OnFailure));
assert_eq!(resolved.backoff().first(), Duration::from_secs(3));
assert_eq!(resolved.backoff().jitter(), JitterPolicy::None);
assert_eq!(resolved.timeout(), Some(Duration::from_secs(12)));
assert_eq!(resolved.max_retries(), Some(retries));
}
#[test]
fn new_does_not_inherit_task_defaults() {
let defaults = TaskDefaults::default()
.with_restart(RestartPolicy::OnFailure)
.with_backoff(BackoffPolicy::constant(Duration::from_secs(8)))
.with_timeout(Duration::from_secs(9))
.with_max_retries(NonZeroU32::new(3).unwrap());
let spec = TaskSpec::new(
task("explicit"),
RestartPolicy::Never,
BackoffPolicy::constant(Duration::from_secs(1)),
None,
);
let resolved = spec.resolve(&defaults);
assert!(matches!(resolved.restart(), RestartPolicy::Never));
assert_eq!(resolved.backoff().first(), Duration::from_secs(1));
assert_eq!(resolved.timeout(), None);
assert_eq!(resolved.max_retries(), None);
}
#[test]
fn with_timeout_accepts_duration_or_option_and_normalizes_zero() {
assert_explicit_timeout(
TaskSpec::once(task("zero-duration")).with_timeout(Duration::ZERO),
None,
"with_timeout(ZERO) must normalize to None",
);
assert_explicit_timeout(
TaskSpec::once(task("zero-option")).with_timeout(Some(Duration::ZERO)),
None,
"with_timeout(Some(ZERO)) must normalize to None",
);
assert_explicit_timeout(
TaskSpec::new(
task("z"),
RestartPolicy::Never,
BackoffPolicy::default(),
Some(Duration::ZERO),
),
None,
"new(.., Some(ZERO)) must normalize to None",
);
let duration = Duration::from_secs(1);
assert_explicit_timeout(
TaskSpec::once(task("positive-duration")).with_timeout(duration),
Some(duration),
"a positive Duration must be preserved",
);
assert_explicit_timeout(
TaskSpec::once(task("positive-option")).with_timeout(Some(duration)),
Some(duration),
"a positive Some(Duration) must be preserved",
);
assert_explicit_timeout(
TaskSpec::once(task("none-inference")).with_timeout(None),
None,
"None must infer Option<Duration> and explicitly disable the timeout",
);
}
#[test]
fn raw_retry_limit_is_validated_like_task_defaults() {
let spec = TaskSpec::once(task("limited"))
.try_with_max_retries(3)
.expect("a positive retry limit must be accepted");
assert!(matches!(
spec.max_retries_override(),
TaskSetting::Explicit(Some(limit)) if limit.get() == 3
));
assert_eq!(
TaskSpec::once(task("zero"))
.try_with_max_retries(0)
.unwrap_err(),
ConfigError::Zero {
field: "max_retries"
}
);
}
}