cronoq-core 0.2.1

//! Task specification and identifiers.

use core::fmt;
use core::time::Duration;
use std::sync::Arc;

use smallvec::SmallVec;

use crate::time::Time;

/// Engine-assigned task identifier (monotonic; the high 16 bits carry the leader epoch).
///
/// With sharding ([`CoreConfig::id_shard`](crate::CoreConfig::id_shard)) the top 8 bits carry
/// the shard number and the leader epoch sits below them.
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct TaskId(pub u64);

impl TaskId {
    /// The shard number of an id issued by a sharded engine (its top 8 bits).
    pub fn shard(self) -> u8 {
        (self.0 >> 56) as u8
    }

    /// The first id a sharded engine issues in `epoch`. Ids then grow by one; the
    /// sequence carries into the epoch bits, never into the shard's.
    pub fn first_in_shard(shard: u8, epoch: u16) -> TaskId {
        TaskId((shard as u64) << 56 | (epoch as u64) << 40 | 1)
    }
}

impl fmt::Display for TaskId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Identifies one dispatch (attempt) of a task.
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct LeaseId(pub u64);

/// Caller-chosen correlation id for submissions.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct Ticket(pub u64);

/// Index of a pool in the engine configuration.
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct PoolId(pub u16);

/// Index of a resource in the engine configuration.
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ResourceId(pub u16);

/// Priority class. Lower discriminant = more important.
///
/// Classes set the base of a task's effective score; waiting time adds a capped aging bonus
/// and deadlines add urgency (see [`PriorityConfig`](crate::PriorityConfig) and
/// [`UrgencyConfig`](crate::UrgencyConfig)).
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[repr(u8)]
pub enum Priority {
    /// Work that must run before anything else (alerts, interactive critical paths).
    Critical = 0,
    /// Latency-sensitive work (interactive requests).
    High = 1,
    /// Regular work (the default).
    #[default]
    Normal = 2,
    /// Background work that runs when nothing more important is waiting.
    Low = 3,
}

impl Priority {
    /// All classes, most important first.
    pub const ALL: [Priority; 4] = [
        Priority::Critical,
        Priority::High,
        Priority::Normal,
        Priority::Low,
    ];
    /// Number of classes.
    pub const COUNT: usize = 4;

    /// Position in [`Priority::ALL`] (0 = `Critical`).
    #[inline]
    pub fn index(self) -> usize {
        self as usize
    }
    /// The class at `i` in [`Priority::ALL`].
    pub fn from_index(i: usize) -> Option<Priority> {
        Self::ALL.get(i).copied()
    }
    /// Lowercase name (`"critical"`, `"high"`, `"normal"`, `"low"`).
    pub fn name(self) -> &'static str {
        match self {
            Priority::Critical => "critical",
            Priority::High => "high",
            Priority::Normal => "normal",
            Priority::Low => "low",
        }
    }
}

/// Reference to another task, used for dependencies and cancellation.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum TaskRef {
    /// By engine-assigned id.
    Id(TaskId),
    /// By the caller-chosen key given at submission.
    Key(Arc<str>),
}

impl From<TaskId> for TaskRef {
    fn from(id: TaskId) -> Self {
        TaskRef::Id(id)
    }
}
impl From<&str> for TaskRef {
    fn from(k: &str) -> Self {
        TaskRef::Key(k.into())
    }
}
impl From<String> for TaskRef {
    fn from(k: String) -> Self {
        TaskRef::Key(k.into())
    }
}

/// When a dependency edge is satisfied by its parent's terminal status.
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
pub enum EdgeCond {
    /// Parent succeeded.
    #[default]
    OnSuccess,
    /// Parent reached any terminal status.
    OnComplete,
    /// Parent failed or was dead-lettered.
    OnFailure,
}

/// A dependency on another task. The dependent becomes eligible once every edge is satisfied
/// and ends as `Skipped` as soon as one edge can no longer be satisfied.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct Dep {
    /// The parent task.
    pub target: TaskRef,
    /// Which parent outcomes satisfy the edge.
    pub cond: EdgeCond,
}

impl Dep {
    /// Run after `target` succeeds.
    pub fn on_success(target: impl Into<TaskRef>) -> Dep {
        Dep {
            target: target.into(),
            cond: EdgeCond::OnSuccess,
        }
    }
    /// Run after `target` ends, whatever its outcome (cleanup steps).
    pub fn on_complete(target: impl Into<TaskRef>) -> Dep {
        Dep {
            target: target.into(),
            cond: EdgeCond::OnComplete,
        }
    }
    /// Run only if `target` fails or is dead-lettered (failure handlers).
    pub fn on_failure(target: impl Into<TaskRef>) -> Dep {
        Dep {
            target: target.into(),
            cond: EdgeCond::OnFailure,
        }
    }
}

/// What to do when a fixed-start task could not start within its tolerance.
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
pub enum MissPolicy {
    /// Start as soon as possible, ahead of all other work.
    #[default]
    RunLate,
    /// Give up; the task ends as `Missed`.
    Skip,
    /// The task ends as `Failed`.
    Fail,
}

/// A start time the scheduler plans capacity for: a worker slot is reserved at admission and
/// held free shortly before `at`.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct FixedStart {
    /// When the task should start.
    pub at: Time,
    /// How late the start may be and still count as on time.
    pub tolerance: Duration,
    /// What happens if the task has not started by `at + tolerance`.
    pub on_miss: MissPolicy,
}

impl FixedStart {
    /// Start at `at` with a 1 s tolerance, running late if missed.
    pub fn at(at: Time) -> FixedStart {
        FixedStart {
            at,
            tolerance: Duration::from_secs(1),
            on_miss: MissPolicy::RunLate,
        }
    }
}

/// Retry behaviour for retryable failures.
#[derive(Copy, Clone, Debug, PartialEq)]
pub struct RetryPolicy {
    /// Total attempts including the first one (>= 1).
    pub max_attempts: u32,
    /// Backoff before the first retry.
    pub initial_backoff: Duration,
    /// Growth factor of the backoff per retry.
    pub multiplier: f64,
    /// Upper bound on the backoff.
    pub max_backoff: Duration,
    /// Full jitter: the delay is uniformly drawn from `[0, backoff]`.
    pub jitter: bool,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        RetryPolicy {
            max_attempts: 3,
            initial_backoff: Duration::from_secs(1),
            multiplier: 2.0,
            max_backoff: Duration::from_secs(60),
            jitter: true,
        }
    }
}

impl RetryPolicy {
    /// A single attempt, never retried.
    pub fn none() -> Self {
        RetryPolicy {
            max_attempts: 1,
            ..Default::default()
        }
    }
    /// Backoff before attempt `next_attempt` (2 = first retry), before jitter.
    pub fn backoff(&self, next_attempt: u32) -> Duration {
        let exp = next_attempt.saturating_sub(2).min(63) as i32;
        let d = self.initial_backoff.as_secs_f64() * self.multiplier.powi(exp);
        if d.is_nan() {
            return self.max_backoff;
        }
        // Saturates: huge or infinite values become `max_backoff` (or `Duration::MAX`).
        let d = Duration::try_from_secs_f64(d.max(0.0)).unwrap_or(Duration::MAX);
        d.min(self.max_backoff)
    }
}

/// Key/value labels attached to a task.
pub type Tags = Arc<[(Arc<str>, Arc<str>)]>;

/// Shared `"default"` name, so building a spec does not allocate for kind and pool.
pub fn default_name() -> Arc<str> {
    static DEFAULT: std::sync::LazyLock<Arc<str>> =
        std::sync::LazyLock::new(|| Arc::from("default"));
    DEFAULT.clone()
}

/// Everything the engine needs to schedule one task.
///
/// Built with [`TaskSpec::new`] and the chained setters; all fields are public for direct use.
#[derive(Debug)]
pub struct TaskSpec<P> {
    /// The work itself, handed to the executor unchanged.
    pub payload: Arc<P>,
    /// Free-form category; duration statistics are learned per `(pool, kind)`.
    pub kind: Arc<str>,
    /// Pool that runs the task (default `"default"`).
    pub pool: Arc<str>,
    /// Priority class.
    pub priority: Priority,
    /// Optional unique key: deduplicates submissions while the task is live or retained, and
    /// lets dependents refer to it by name.
    pub key: Option<Arc<str>>,
    /// Soft deadline for completion; urgency rises as its latest start time approaches.
    pub deadline: Option<Time>,
    /// Start at (or close to) a given time, with capacity reserved in advance.
    pub fixed_start: Option<FixedStart>,
    /// Expected runtime; when absent the engine uses learned statistics.
    pub duration_hint: Option<Duration>,
    /// Per-attempt timeout.
    pub timeout: Option<Duration>,
    /// `None` uses the pool default.
    pub retry: Option<RetryPolicy>,
    /// Tasks that must finish first (see [`Dep`]).
    pub deps: Vec<Dep>,
    /// Shared resources needed while running: `(resource name, amount)`.
    pub resources: SmallVec<[(Arc<str>, u32); 2]>,
    /// Labels passed through to executors (e.g. `traceparent`, tenant).
    pub tags: Vec<(Arc<str>, Arc<str>)>,
    /// Worker labels this task requires: it runs only on a worker group whose labels include
    /// every `(name, value)` pair (see [`Core::set_group_capacity`](crate::Core::set_group_capacity)).
    pub labels: Vec<(Arc<str>, Arc<str>)>,
    /// Tenant the task belongs to, for weighted fair sharing of pools with
    /// [`PoolConfig::fairness`](crate::PoolConfig::fairness).
    pub tenant: Option<Arc<str>>,
}

impl<P> Clone for TaskSpec<P> {
    fn clone(&self) -> Self {
        TaskSpec {
            payload: self.payload.clone(),
            kind: self.kind.clone(),
            pool: self.pool.clone(),
            priority: self.priority,
            key: self.key.clone(),
            deadline: self.deadline,
            fixed_start: self.fixed_start,
            duration_hint: self.duration_hint,
            timeout: self.timeout,
            retry: self.retry,
            deps: self.deps.clone(),
            resources: self.resources.clone(),
            tags: self.tags.clone(),
            labels: self.labels.clone(),
            tenant: self.tenant.clone(),
        }
    }
}

impl<P> TaskSpec<P> {
    /// Same task description with a different payload type.
    pub fn map_payload<Q>(self, payload: Arc<Q>) -> TaskSpec<Q> {
        TaskSpec {
            payload,
            kind: self.kind,
            pool: self.pool,
            priority: self.priority,
            key: self.key,
            deadline: self.deadline,
            fixed_start: self.fixed_start,
            duration_hint: self.duration_hint,
            timeout: self.timeout,
            retry: self.retry,
            deps: self.deps,
            resources: self.resources,
            tags: self.tags,
            labels: self.labels,
            tenant: self.tenant,
        }
    }

    /// A `Normal` task in the `"default"` pool.
    pub fn new(payload: P) -> Self {
        Self::from_arc(Arc::new(payload))
    }
    /// Like [`TaskSpec::new`] with an already shared payload.
    pub fn from_arc(payload: Arc<P>) -> Self {
        TaskSpec {
            payload,
            kind: default_name(),
            pool: default_name(),
            priority: Priority::Normal,
            key: None,
            deadline: None,
            fixed_start: None,
            duration_hint: None,
            timeout: None,
            retry: None,
            deps: Vec::new(),
            resources: SmallVec::new(),
            tags: Vec::new(),
            labels: Vec::new(),
            tenant: None,
        }
    }
    /// Set the kind.
    pub fn kind(mut self, k: impl Into<Arc<str>>) -> Self {
        self.kind = k.into();
        self
    }
    /// Set the pool.
    pub fn pool(mut self, p: impl Into<Arc<str>>) -> Self {
        self.pool = p.into();
        self
    }
    /// Set the priority class.
    pub fn priority(mut self, p: Priority) -> Self {
        self.priority = p;
        self
    }
    /// Set the unique key.
    pub fn key(mut self, k: impl Into<Arc<str>>) -> Self {
        self.key = Some(k.into());
        self
    }
    /// Set the deadline.
    pub fn deadline(mut self, t: Time) -> Self {
        self.deadline = Some(t);
        self
    }
    /// Set a fixed start.
    pub fn fixed_start(mut self, f: FixedStart) -> Self {
        self.fixed_start = Some(f);
        self
    }
    /// Set the expected runtime.
    pub fn duration_hint(mut self, d: Duration) -> Self {
        self.duration_hint = Some(d);
        self
    }
    /// Set the per-attempt timeout.
    pub fn timeout(mut self, d: Duration) -> Self {
        self.timeout = Some(d);
        self
    }
    /// Set the retry policy.
    pub fn retry(mut self, r: RetryPolicy) -> Self {
        self.retry = Some(r);
        self
    }
    /// Add a dependency.
    pub fn dep(mut self, d: Dep) -> Self {
        self.deps.push(d);
        self
    }
    /// Run after `target` succeeds.
    pub fn after(self, target: impl Into<TaskRef>) -> Self {
        self.dep(Dep::on_success(target))
    }
    /// Require `amount` units of a resource while running.
    pub fn resource(mut self, name: impl Into<Arc<str>>, amount: u32) -> Self {
        self.resources.push((name.into(), amount));
        self
    }
    /// Add a label.
    pub fn tag(mut self, k: impl Into<Arc<str>>, v: impl Into<Arc<str>>) -> Self {
        self.tags.push((k.into(), v.into()));
        self
    }
    /// The tenant the task belongs to (for fair sharing).
    pub fn tenant(mut self, tenant: impl Into<Arc<str>>) -> Self {
        self.tenant = Some(tenant.into());
        self
    }
    /// Require a worker label (`name = value`) to run.
    pub fn label(mut self, name: impl Into<Arc<str>>, value: impl Into<Arc<str>>) -> Self {
        self.labels.push((name.into(), value.into()));
        self
    }
}