cronoq_core/
spec.rs

1//! Task specification and identifiers.
2
3use core::fmt;
4use core::time::Duration;
5use std::sync::Arc;
6
7use smallvec::SmallVec;
8
9use crate::time::Time;
10
11/// Engine-assigned task identifier (monotonic; the high 16 bits carry the leader epoch).
12///
13/// With sharding ([`CoreConfig::id_shard`](crate::CoreConfig::id_shard)) the top 8 bits carry
14/// the shard number and the leader epoch sits below them.
15#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
16pub struct TaskId(pub u64);
17
18impl TaskId {
19    /// The shard number of an id issued by a sharded engine (its top 8 bits).
20    pub fn shard(self) -> u8 {
21        (self.0 >> 56) as u8
22    }
23
24    /// The first id a sharded engine issues in `epoch`. Ids then grow by one; the
25    /// sequence carries into the epoch bits, never into the shard's.
26    pub fn first_in_shard(shard: u8, epoch: u16) -> TaskId {
27        TaskId((shard as u64) << 56 | (epoch as u64) << 40 | 1)
28    }
29}
30
31impl fmt::Display for TaskId {
32    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33        write!(f, "{}", self.0)
34    }
35}
36
37/// Identifies one dispatch (attempt) of a task.
38#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
39pub struct LeaseId(pub u64);
40
41/// Caller-chosen correlation id for submissions.
42#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
43pub struct Ticket(pub u64);
44
45/// Index of a pool in the engine configuration.
46#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
47pub struct PoolId(pub u16);
48
49/// Index of a resource in the engine configuration.
50#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
51pub struct ResourceId(pub u16);
52
53/// Priority class. Lower discriminant = more important.
54///
55/// Classes set the base of a task's effective score; waiting time adds a capped aging bonus
56/// and deadlines add urgency (see [`PriorityConfig`](crate::PriorityConfig) and
57/// [`UrgencyConfig`](crate::UrgencyConfig)).
58#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
59#[repr(u8)]
60pub enum Priority {
61    /// Work that must run before anything else (alerts, interactive critical paths).
62    Critical = 0,
63    /// Latency-sensitive work (interactive requests).
64    High = 1,
65    /// Regular work (the default).
66    #[default]
67    Normal = 2,
68    /// Background work that runs when nothing more important is waiting.
69    Low = 3,
70}
71
72impl Priority {
73    /// All classes, most important first.
74    pub const ALL: [Priority; 4] = [
75        Priority::Critical,
76        Priority::High,
77        Priority::Normal,
78        Priority::Low,
79    ];
80    /// Number of classes.
81    pub const COUNT: usize = 4;
82
83    /// Position in [`Priority::ALL`] (0 = `Critical`).
84    #[inline]
85    pub fn index(self) -> usize {
86        self as usize
87    }
88    /// The class at `i` in [`Priority::ALL`].
89    pub fn from_index(i: usize) -> Option<Priority> {
90        Self::ALL.get(i).copied()
91    }
92    /// Lowercase name (`"critical"`, `"high"`, `"normal"`, `"low"`).
93    pub fn name(self) -> &'static str {
94        match self {
95            Priority::Critical => "critical",
96            Priority::High => "high",
97            Priority::Normal => "normal",
98            Priority::Low => "low",
99        }
100    }
101}
102
103/// Reference to another task, used for dependencies and cancellation.
104#[derive(Clone, Debug, PartialEq, Eq, Hash)]
105pub enum TaskRef {
106    /// By engine-assigned id.
107    Id(TaskId),
108    /// By the caller-chosen key given at submission.
109    Key(Arc<str>),
110}
111
112impl From<TaskId> for TaskRef {
113    fn from(id: TaskId) -> Self {
114        TaskRef::Id(id)
115    }
116}
117impl From<&str> for TaskRef {
118    fn from(k: &str) -> Self {
119        TaskRef::Key(k.into())
120    }
121}
122impl From<String> for TaskRef {
123    fn from(k: String) -> Self {
124        TaskRef::Key(k.into())
125    }
126}
127
128/// When a dependency edge is satisfied by its parent's terminal status.
129#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
130pub enum EdgeCond {
131    /// Parent succeeded.
132    #[default]
133    OnSuccess,
134    /// Parent reached any terminal status.
135    OnComplete,
136    /// Parent failed or was dead-lettered.
137    OnFailure,
138}
139
140/// A dependency on another task. The dependent becomes eligible once every edge is satisfied
141/// and ends as `Skipped` as soon as one edge can no longer be satisfied.
142#[derive(Clone, Debug, PartialEq, Eq, Hash)]
143pub struct Dep {
144    /// The parent task.
145    pub target: TaskRef,
146    /// Which parent outcomes satisfy the edge.
147    pub cond: EdgeCond,
148}
149
150impl Dep {
151    /// Run after `target` succeeds.
152    pub fn on_success(target: impl Into<TaskRef>) -> Dep {
153        Dep {
154            target: target.into(),
155            cond: EdgeCond::OnSuccess,
156        }
157    }
158    /// Run after `target` ends, whatever its outcome (cleanup steps).
159    pub fn on_complete(target: impl Into<TaskRef>) -> Dep {
160        Dep {
161            target: target.into(),
162            cond: EdgeCond::OnComplete,
163        }
164    }
165    /// Run only if `target` fails or is dead-lettered (failure handlers).
166    pub fn on_failure(target: impl Into<TaskRef>) -> Dep {
167        Dep {
168            target: target.into(),
169            cond: EdgeCond::OnFailure,
170        }
171    }
172}
173
174/// What to do when a fixed-start task could not start within its tolerance.
175#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
176pub enum MissPolicy {
177    /// Start as soon as possible, ahead of all other work.
178    #[default]
179    RunLate,
180    /// Give up; the task ends as `Missed`.
181    Skip,
182    /// The task ends as `Failed`.
183    Fail,
184}
185
186/// A start time the scheduler plans capacity for: a worker slot is reserved at admission and
187/// held free shortly before `at`.
188#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
189pub struct FixedStart {
190    /// When the task should start.
191    pub at: Time,
192    /// How late the start may be and still count as on time.
193    pub tolerance: Duration,
194    /// What happens if the task has not started by `at + tolerance`.
195    pub on_miss: MissPolicy,
196}
197
198impl FixedStart {
199    /// Start at `at` with a 1 s tolerance, running late if missed.
200    pub fn at(at: Time) -> FixedStart {
201        FixedStart {
202            at,
203            tolerance: Duration::from_secs(1),
204            on_miss: MissPolicy::RunLate,
205        }
206    }
207}
208
209/// Retry behaviour for retryable failures.
210#[derive(Copy, Clone, Debug, PartialEq)]
211pub struct RetryPolicy {
212    /// Total attempts including the first one (>= 1).
213    pub max_attempts: u32,
214    /// Backoff before the first retry.
215    pub initial_backoff: Duration,
216    /// Growth factor of the backoff per retry.
217    pub multiplier: f64,
218    /// Upper bound on the backoff.
219    pub max_backoff: Duration,
220    /// Full jitter: the delay is uniformly drawn from `[0, backoff]`.
221    pub jitter: bool,
222}
223
224impl Default for RetryPolicy {
225    fn default() -> Self {
226        RetryPolicy {
227            max_attempts: 3,
228            initial_backoff: Duration::from_secs(1),
229            multiplier: 2.0,
230            max_backoff: Duration::from_secs(60),
231            jitter: true,
232        }
233    }
234}
235
236impl RetryPolicy {
237    /// A single attempt, never retried.
238    pub fn none() -> Self {
239        RetryPolicy {
240            max_attempts: 1,
241            ..Default::default()
242        }
243    }
244    /// Backoff before attempt `next_attempt` (2 = first retry), before jitter.
245    pub fn backoff(&self, next_attempt: u32) -> Duration {
246        let exp = next_attempt.saturating_sub(2).min(63) as i32;
247        let d = self.initial_backoff.as_secs_f64() * self.multiplier.powi(exp);
248        if d.is_nan() {
249            return self.max_backoff;
250        }
251        // Saturates: huge or infinite values become `max_backoff` (or `Duration::MAX`).
252        let d = Duration::try_from_secs_f64(d.max(0.0)).unwrap_or(Duration::MAX);
253        d.min(self.max_backoff)
254    }
255}
256
257/// Key/value labels attached to a task.
258pub type Tags = Arc<[(Arc<str>, Arc<str>)]>;
259
260/// Shared `"default"` name, so building a spec does not allocate for kind and pool.
261pub fn default_name() -> Arc<str> {
262    static DEFAULT: std::sync::LazyLock<Arc<str>> =
263        std::sync::LazyLock::new(|| Arc::from("default"));
264    DEFAULT.clone()
265}
266
267/// Everything the engine needs to schedule one task.
268///
269/// Built with [`TaskSpec::new`] and the chained setters; all fields are public for direct use.
270#[derive(Debug)]
271pub struct TaskSpec<P> {
272    /// The work itself, handed to the executor unchanged.
273    pub payload: Arc<P>,
274    /// Free-form category; duration statistics are learned per `(pool, kind)`.
275    pub kind: Arc<str>,
276    /// Pool that runs the task (default `"default"`).
277    pub pool: Arc<str>,
278    /// Priority class.
279    pub priority: Priority,
280    /// Optional unique key: deduplicates submissions while the task is live or retained, and
281    /// lets dependents refer to it by name.
282    pub key: Option<Arc<str>>,
283    /// Soft deadline for completion; urgency rises as its latest start time approaches.
284    pub deadline: Option<Time>,
285    /// Start at (or close to) a given time, with capacity reserved in advance.
286    pub fixed_start: Option<FixedStart>,
287    /// Expected runtime; when absent the engine uses learned statistics.
288    pub duration_hint: Option<Duration>,
289    /// Per-attempt timeout.
290    pub timeout: Option<Duration>,
291    /// `None` uses the pool default.
292    pub retry: Option<RetryPolicy>,
293    /// Tasks that must finish first (see [`Dep`]).
294    pub deps: Vec<Dep>,
295    /// Shared resources needed while running: `(resource name, amount)`.
296    pub resources: SmallVec<[(Arc<str>, u32); 2]>,
297    /// Labels passed through to executors (e.g. `traceparent`, tenant).
298    pub tags: Vec<(Arc<str>, Arc<str>)>,
299    /// Worker labels this task requires: it runs only on a worker group whose labels include
300    /// every `(name, value)` pair (see [`Core::set_group_capacity`](crate::Core::set_group_capacity)).
301    pub labels: Vec<(Arc<str>, Arc<str>)>,
302    /// Tenant the task belongs to, for weighted fair sharing of pools with
303    /// [`PoolConfig::fairness`](crate::PoolConfig::fairness).
304    pub tenant: Option<Arc<str>>,
305}
306
307impl<P> Clone for TaskSpec<P> {
308    fn clone(&self) -> Self {
309        TaskSpec {
310            payload: self.payload.clone(),
311            kind: self.kind.clone(),
312            pool: self.pool.clone(),
313            priority: self.priority,
314            key: self.key.clone(),
315            deadline: self.deadline,
316            fixed_start: self.fixed_start,
317            duration_hint: self.duration_hint,
318            timeout: self.timeout,
319            retry: self.retry,
320            deps: self.deps.clone(),
321            resources: self.resources.clone(),
322            tags: self.tags.clone(),
323            labels: self.labels.clone(),
324            tenant: self.tenant.clone(),
325        }
326    }
327}
328
329impl<P> TaskSpec<P> {
330    /// Same task description with a different payload type.
331    pub fn map_payload<Q>(self, payload: Arc<Q>) -> TaskSpec<Q> {
332        TaskSpec {
333            payload,
334            kind: self.kind,
335            pool: self.pool,
336            priority: self.priority,
337            key: self.key,
338            deadline: self.deadline,
339            fixed_start: self.fixed_start,
340            duration_hint: self.duration_hint,
341            timeout: self.timeout,
342            retry: self.retry,
343            deps: self.deps,
344            resources: self.resources,
345            tags: self.tags,
346            labels: self.labels,
347            tenant: self.tenant,
348        }
349    }
350
351    /// A `Normal` task in the `"default"` pool.
352    pub fn new(payload: P) -> Self {
353        Self::from_arc(Arc::new(payload))
354    }
355    /// Like [`TaskSpec::new`] with an already shared payload.
356    pub fn from_arc(payload: Arc<P>) -> Self {
357        TaskSpec {
358            payload,
359            kind: default_name(),
360            pool: default_name(),
361            priority: Priority::Normal,
362            key: None,
363            deadline: None,
364            fixed_start: None,
365            duration_hint: None,
366            timeout: None,
367            retry: None,
368            deps: Vec::new(),
369            resources: SmallVec::new(),
370            tags: Vec::new(),
371            labels: Vec::new(),
372            tenant: None,
373        }
374    }
375    /// Set the kind.
376    pub fn kind(mut self, k: impl Into<Arc<str>>) -> Self {
377        self.kind = k.into();
378        self
379    }
380    /// Set the pool.
381    pub fn pool(mut self, p: impl Into<Arc<str>>) -> Self {
382        self.pool = p.into();
383        self
384    }
385    /// Set the priority class.
386    pub fn priority(mut self, p: Priority) -> Self {
387        self.priority = p;
388        self
389    }
390    /// Set the unique key.
391    pub fn key(mut self, k: impl Into<Arc<str>>) -> Self {
392        self.key = Some(k.into());
393        self
394    }
395    /// Set the deadline.
396    pub fn deadline(mut self, t: Time) -> Self {
397        self.deadline = Some(t);
398        self
399    }
400    /// Set a fixed start.
401    pub fn fixed_start(mut self, f: FixedStart) -> Self {
402        self.fixed_start = Some(f);
403        self
404    }
405    /// Set the expected runtime.
406    pub fn duration_hint(mut self, d: Duration) -> Self {
407        self.duration_hint = Some(d);
408        self
409    }
410    /// Set the per-attempt timeout.
411    pub fn timeout(mut self, d: Duration) -> Self {
412        self.timeout = Some(d);
413        self
414    }
415    /// Set the retry policy.
416    pub fn retry(mut self, r: RetryPolicy) -> Self {
417        self.retry = Some(r);
418        self
419    }
420    /// Add a dependency.
421    pub fn dep(mut self, d: Dep) -> Self {
422        self.deps.push(d);
423        self
424    }
425    /// Run after `target` succeeds.
426    pub fn after(self, target: impl Into<TaskRef>) -> Self {
427        self.dep(Dep::on_success(target))
428    }
429    /// Require `amount` units of a resource while running.
430    pub fn resource(mut self, name: impl Into<Arc<str>>, amount: u32) -> Self {
431        self.resources.push((name.into(), amount));
432        self
433    }
434    /// Add a label.
435    pub fn tag(mut self, k: impl Into<Arc<str>>, v: impl Into<Arc<str>>) -> Self {
436        self.tags.push((k.into(), v.into()));
437        self
438    }
439    /// The tenant the task belongs to (for fair sharing).
440    pub fn tenant(mut self, tenant: impl Into<Arc<str>>) -> Self {
441        self.tenant = Some(tenant.into());
442        self
443    }
444    /// Require a worker label (`name = value`) to run.
445    pub fn label(mut self, name: impl Into<Arc<str>>, value: impl Into<Arc<str>>) -> Self {
446        self.labels.push((name.into(), value.into()));
447        self
448    }
449}
cronoq_core/spec.rs

cronoq_core/
spec.rs
