cronoq_core/io.rs
1//! Engine outputs (actions), outcomes, statuses, errors and persistence mutations.
2
3use core::fmt;
4use core::time::Duration;
5use std::sync::Arc;
6
7use crate::spec::{LeaseId, MissPolicy, PoolId, Priority, TaskId, TaskSpec};
8use crate::time::Time;
9
10/// Lifecycle status of a task.
11#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
12pub enum TaskStatus {
13 /// Waiting for dependencies.
14 Blocked,
15 /// Waiting for its fixed start time.
16 Scheduled,
17 /// Eligible and queued.
18 Ready,
19 /// An attempt is running on a worker.
20 Running,
21 /// Waiting for a retry backoff to elapse.
22 Retrying,
23 /// Completed successfully.
24 Succeeded,
25 /// Permanent failure (or a fixed-start miss with `MissPolicy::Fail`).
26 Failed,
27 /// Retryable failures exhausted all attempts.
28 DeadLettered,
29 /// Cancelled by a caller.
30 Cancelled,
31 /// A dependency edge condition can no longer be met.
32 Skipped,
33 /// Fixed start missed with `MissPolicy::Skip`.
34 Missed,
35 /// Evicted under overload.
36 Shed,
37}
38
39impl TaskStatus {
40 /// Whether the task has finished for good.
41 pub fn is_terminal(self) -> bool {
42 !matches!(
43 self,
44 TaskStatus::Blocked
45 | TaskStatus::Scheduled
46 | TaskStatus::Ready
47 | TaskStatus::Running
48 | TaskStatus::Retrying
49 )
50 }
51 /// Stable lower-case name (used in storage and APIs).
52 pub fn name(self) -> &'static str {
53 match self {
54 TaskStatus::Blocked => "blocked",
55 TaskStatus::Scheduled => "scheduled",
56 TaskStatus::Ready => "ready",
57 TaskStatus::Running => "running",
58 TaskStatus::Retrying => "retrying",
59 TaskStatus::Succeeded => "succeeded",
60 TaskStatus::Failed => "failed",
61 TaskStatus::DeadLettered => "dead_lettered",
62 TaskStatus::Cancelled => "cancelled",
63 TaskStatus::Skipped => "skipped",
64 TaskStatus::Missed => "missed",
65 TaskStatus::Shed => "shed",
66 }
67 }
68 /// Inverse of [`TaskStatus::name`].
69 pub fn from_name(s: &str) -> Option<TaskStatus> {
70 use TaskStatus::*;
71 Some(match s {
72 "blocked" => Blocked,
73 "scheduled" => Scheduled,
74 "ready" => Ready,
75 "running" => Running,
76 "retrying" => Retrying,
77 "succeeded" => Succeeded,
78 "failed" => Failed,
79 "dead_lettered" => DeadLettered,
80 "cancelled" => Cancelled,
81 "skipped" => Skipped,
82 "missed" => Missed,
83 "shed" => Shed,
84 _ => return None,
85 })
86 }
87}
88
89impl fmt::Display for TaskStatus {
90 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
91 f.write_str(self.name())
92 }
93}
94
95/// Result of one attempt, reported by the executor.
96#[derive(Clone, Debug, PartialEq)]
97pub enum Outcome {
98 /// The attempt succeeded.
99 Success,
100 /// The attempt failed.
101 Failure {
102 /// Whether the failure may be retried under the task's retry policy.
103 retryable: bool,
104 /// Minimum delay before the retry (overrides a shorter backoff).
105 retry_after: Option<Duration>,
106 /// Error description.
107 message: Arc<str>,
108 },
109 /// The executor observed cancellation.
110 Cancelled,
111}
112
113impl Outcome {
114 /// A retryable failure.
115 pub fn retryable(msg: impl Into<Arc<str>>) -> Outcome {
116 Outcome::Failure {
117 retryable: true,
118 retry_after: None,
119 message: msg.into(),
120 }
121 }
122 /// A permanent (non-retryable) failure.
123 pub fn permanent(msg: impl Into<Arc<str>>) -> Outcome {
124 Outcome::Failure {
125 retryable: false,
126 retry_after: None,
127 message: msg.into(),
128 }
129 }
130}
131
132/// Why a submission was rejected.
133#[derive(Clone, Debug, PartialEq, Eq)]
134pub enum SubmitError {
135 /// No pool with this name.
136 UnknownPool(Arc<str>),
137 /// No resource with this name.
138 UnknownResource(Arc<str>),
139 /// Requested amount can never be satisfied by the resource.
140 ResourceExceedsCapacity(Arc<str>),
141 /// The pool's fixed-start reservation limit would be exceeded at that time.
142 ReservationConflict,
143 /// The dependencies would form a cycle.
144 Cycle,
145 /// A dependency names an unknown key and the pool rejects unknown dependencies.
146 UnknownDependency(Arc<str>),
147 /// Bounds reached and the pool's policy is `Wait`: retry once `Action::SpaceAvailable` is seen.
148 Full,
149 /// Bounds reached and the policy is `Reject` (or shedding found no victim).
150 Overloaded,
151 /// The key already exists and the batch tried to define it twice.
152 DuplicateKeyInBatch(Arc<str>),
153 /// The spec is malformed.
154 Invalid(&'static str),
155}
156
157impl fmt::Display for SubmitError {
158 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
159 match self {
160 SubmitError::UnknownPool(p) => write!(f, "unknown pool '{p}'"),
161 SubmitError::UnknownResource(r) => write!(f, "unknown resource '{r}'"),
162 SubmitError::ResourceExceedsCapacity(r) => {
163 write!(f, "requested amount exceeds capacity of resource '{r}'")
164 }
165 SubmitError::ReservationConflict => f.write_str("fixed-start reservation conflict"),
166 SubmitError::Cycle => f.write_str("dependency cycle"),
167 SubmitError::UnknownDependency(k) => write!(f, "unknown dependency '{k}'"),
168 SubmitError::Full => f.write_str("queue full"),
169 SubmitError::Overloaded => f.write_str("overloaded"),
170 SubmitError::DuplicateKeyInBatch(k) => write!(f, "duplicate key '{k}' in batch"),
171 SubmitError::Invalid(m) => write!(f, "invalid task: {m}"),
172 }
173 }
174}
175
176impl std::error::Error for SubmitError {}
177
178/// Result of an accepted submission.
179#[derive(Copy, Clone, Debug, PartialEq, Eq)]
180pub struct Submitted {
181 /// Id of the task.
182 pub id: TaskId,
183 /// The key was already live; `id` is the existing task.
184 pub duplicate: bool,
185}
186
187/// Why the engine aborted an attempt.
188#[derive(Copy, Clone, Debug, PartialEq, Eq)]
189pub enum AbortReason {
190 /// The attempt exceeded the task's timeout.
191 Timeout,
192 /// The task was cancelled.
193 Cancelled,
194 /// The worker stopped heartbeating its lease.
195 LeaseExpired,
196}
197
198/// A task handed to an executor.
199#[derive(Debug)]
200pub struct Dispatch<P> {
201 /// Lease identifying this attempt; pass it back to `complete` and `heartbeat`.
202 pub lease: LeaseId,
203 /// Task id.
204 pub id: TaskId,
205 /// Pool the task runs in.
206 pub pool: PoolId,
207 /// Attempt number, starting at 1.
208 pub attempt: u32,
209 /// Opaque task payload.
210 pub payload: Arc<P>,
211 /// Task kind (selects the handler).
212 pub kind: Arc<str>,
213 /// Idempotency key, if any.
214 pub key: Option<Arc<str>>,
215 /// Per-attempt timeout, if any.
216 pub timeout: Option<Duration>,
217 /// Completion deadline, if any.
218 pub deadline: Option<Time>,
219 /// Task tags.
220 pub tags: crate::spec::Tags,
221 /// Labels of the worker group chosen for this attempt (empty for the unlabeled group);
222 /// a driver with several groups must run the attempt on a worker of this group.
223 pub group: Labels,
224}
225
226/// Sorted `(name, value)` labels of a worker group.
227pub type Labels = Arc<[(Arc<str>, Arc<str>)]>;
228
229impl<P> Clone for Dispatch<P> {
230 fn clone(&self) -> Self {
231 Dispatch {
232 lease: self.lease,
233 id: self.id,
234 pool: self.pool,
235 attempt: self.attempt,
236 payload: self.payload.clone(),
237 kind: self.kind.clone(),
238 key: self.key.clone(),
239 timeout: self.timeout,
240 deadline: self.deadline,
241 tags: self.tags.clone(),
242 group: self.group.clone(),
243 }
244 }
245}
246
247/// Status transition event.
248#[derive(Clone, Debug, PartialEq)]
249pub struct StatusEvent {
250 /// Task id.
251 pub id: TaskId,
252 /// Pool of the task.
253 pub pool: PoolId,
254 /// Idempotency key, if any.
255 pub key: Option<Arc<str>>,
256 /// New status.
257 pub status: TaskStatus,
258 /// Attempts made so far.
259 pub attempt: u32,
260 /// Engine time of the transition.
261 pub at: Time,
262 /// Error or reason, if any.
263 pub message: Option<Arc<str>>,
264}
265
266/// Overlap behaviour for recurring schedules.
267#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
268pub enum OverlapPolicy {
269 /// Skip an occurrence while the previous one is still live.
270 #[default]
271 Skip,
272 /// Start occurrences regardless of the previous one.
273 Allow,
274 /// Queue each occurrence behind the previous one: it starts once the previous one has
275 /// finished, whatever the outcome.
276 Queue,
277 /// Queue each occurrence behind the previous one and run it only if the previous one
278 /// succeeded; otherwise it ends as `Skipped`, and so do the following ones, until the
279 /// schedule is added again.
280 QueueOnSuccess,
281}
282
283/// A cron schedule that submits a fixed-start task from `template` at each occurrence.
284#[derive(Debug)]
285pub struct RecurringSpec<P> {
286 /// Unique schedule name (occurrences are tagged `cronoq.recurring=<name>`).
287 pub name: Arc<str>,
288 /// Cron expression (5 or 6 fields, seconds optional).
289 pub cron: Arc<str>,
290 /// IANA time zone (default UTC).
291 pub tz: Option<Arc<str>>,
292 /// Spec of each occurrence. Its `fixed_start` is set to the occurrence; a `key` is a
293 /// prefix that each occurrence extends with `@` and its UTC start time (RFC 3339).
294 pub template: TaskSpec<P>,
295 /// Deadline of each occurrence, relative to its start.
296 pub deadline_after: Option<Duration>,
297 /// What to do while the previous occurrence is still live.
298 pub overlap: OverlapPolicy,
299 /// How late an occurrence may start.
300 pub tolerance: Duration,
301 /// What to do when an occurrence misses its tolerance.
302 pub on_miss: MissPolicy,
303}
304
305impl<P> Clone for RecurringSpec<P> {
306 fn clone(&self) -> Self {
307 RecurringSpec {
308 name: self.name.clone(),
309 cron: self.cron.clone(),
310 tz: self.tz.clone(),
311 template: self.template.clone(),
312 deadline_after: self.deadline_after,
313 overlap: self.overlap,
314 tolerance: self.tolerance,
315 on_miss: self.on_miss,
316 }
317 }
318}
319
320/// Durable state change (emitted only with `CoreConfig::persist`).
321#[derive(Debug)]
322pub enum Mutation<P> {
323 /// A task was accepted.
324 Submitted {
325 /// Task id.
326 id: TaskId,
327 /// Submission sequence number (FIFO tie-break).
328 seq: u64,
329 /// Submission time.
330 arrival: Time,
331 /// The task spec.
332 spec: TaskSpec<P>,
333 },
334 /// A task changed status.
335 Status {
336 /// Task id.
337 id: TaskId,
338 /// New status.
339 status: TaskStatus,
340 /// Attempts made so far.
341 attempt: u32,
342 /// Time of the transition.
343 at: Time,
344 /// Error or reason, if any.
345 message: Option<Arc<str>>,
346 },
347 /// A recurring schedule was added or replaced.
348 RecurringUpsert {
349 /// The schedule.
350 spec: RecurringSpec<P>,
351 },
352 /// A recurring schedule was removed.
353 RecurringRemove {
354 /// Schedule name.
355 name: Arc<str>,
356 },
357 /// Next occurrence time of a recurring schedule.
358 RecurringNext {
359 /// Schedule name.
360 name: Arc<str>,
361 /// Next occurrence (`None` = the schedule has ended).
362 next: Option<Time>,
363 /// Task of the latest occurrence a queued schedule chains to (`None` = no chain yet).
364 last: Option<TaskId>,
365 },
366}
367
368impl<P> Clone for Mutation<P> {
369 fn clone(&self) -> Self {
370 match self {
371 Mutation::Submitted {
372 id,
373 seq,
374 arrival,
375 spec,
376 } => Mutation::Submitted {
377 id: *id,
378 seq: *seq,
379 arrival: *arrival,
380 spec: spec.clone(),
381 },
382 Mutation::Status {
383 id,
384 status,
385 attempt,
386 at,
387 message,
388 } => Mutation::Status {
389 id: *id,
390 status: *status,
391 attempt: *attempt,
392 at: *at,
393 message: message.clone(),
394 },
395 Mutation::RecurringUpsert { spec } => Mutation::RecurringUpsert { spec: spec.clone() },
396 Mutation::RecurringRemove { name } => Mutation::RecurringRemove { name: name.clone() },
397 Mutation::RecurringNext { name, next, last } => Mutation::RecurringNext {
398 name: name.clone(),
399 next: *next,
400 last: *last,
401 },
402 }
403 }
404}
405
406/// Something the driver must do.
407#[derive(Debug)]
408pub enum Action<P> {
409 /// Run an attempt on a free worker of its pool.
410 Dispatch(Dispatch<P>),
411 /// Stop the attempt (the engine already treats it as finished).
412 Abort {
413 /// Lease of the attempt.
414 lease: LeaseId,
415 /// Task id.
416 id: TaskId,
417 /// Why it was aborted.
418 reason: AbortReason,
419 },
420 /// Ask the pool's scaler for `desired` workers; report back with `set_capacity`.
421 ScaleTo {
422 /// Pool to scale.
423 pool: PoolId,
424 /// Requested worker count.
425 desired: u32,
426 },
427 /// A task changed status.
428 Status(StatusEvent),
429 /// Write this mutation to durable storage (only with `CoreConfig::persist`).
430 Persist(Box<Mutation<P>>),
431 /// Space freed after a `SubmitError::Full`; waiting submitters may retry.
432 SpaceAvailable,
433}
434
435/// A non-terminal task restored from durable storage.
436#[derive(Debug)]
437pub struct RecoveredTask<P> {
438 /// Task id.
439 pub id: TaskId,
440 /// Original submission sequence number.
441 pub seq: u64,
442 /// Original submission time.
443 pub arrival: Time,
444 /// The task spec.
445 pub spec: TaskSpec<P>,
446 /// Attempts already made (a task that was running counts its lost attempt).
447 pub attempt: u32,
448}
449
450/// A recently finished task restored into the retention table (for late dependents).
451#[derive(Clone, Debug)]
452pub struct RecoveredTerminal {
453 /// Task id.
454 pub id: TaskId,
455 /// Idempotency key, if any.
456 pub key: Option<Arc<str>>,
457 /// Terminal status.
458 pub status: TaskStatus,
459 /// When it finished.
460 pub at: Time,
461}
462
463/// A recurring schedule restored from durable storage.
464#[derive(Debug)]
465pub struct RecoveredRecurring<P> {
466 /// The schedule.
467 pub spec: RecurringSpec<P>,
468 /// Its stored next occurrence.
469 pub next: Option<Time>,
470 /// Task of the latest occurrence a queued schedule chains to.
471 pub last: Option<TaskId>,
472}
473
474/// Engine state restored from durable storage, passed to `Core::recover`.
475#[derive(Debug)]
476pub struct Recovered<P> {
477 /// Live (non-terminal) tasks.
478 pub tasks: Vec<RecoveredTask<P>>,
479 /// Recently finished tasks.
480 pub terminal: Vec<RecoveredTerminal>,
481 /// Recurring schedules.
482 pub recurring: Vec<RecoveredRecurring<P>>,
483 /// Lowest task id the engine may hand out next.
484 pub next_id: u64,
485}
486
487impl<P> Recovered<P> {
488 /// The first recovered task id that does not belong to `shard`, if any: the store holds
489 /// tasks of another shard. Ids of an unsharded server are caught only when their top byte
490 /// differs from `shard` (early epochs look like shard 0's), so give shards new stores.
491 pub fn outside_shard(&self, shard: u8) -> Option<TaskId> {
492 let last = (self.next_id > 0).then(|| TaskId(self.next_id - 1));
493 last.into_iter()
494 .chain(self.tasks.iter().map(|t| t.id))
495 .chain(self.terminal.iter().map(|t| t.id))
496 .find(|id| id.shard() != shard)
497 }
498}
499
500impl<P> Default for Recovered<P> {
501 fn default() -> Self {
502 Recovered {
503 tasks: Vec::new(),
504 terminal: Vec::new(),
505 recurring: Vec::new(),
506 next_id: 0,
507 }
508 }
509}
510
511/// A live task, as listed by [`Core::list_tasks`](crate::Core::list_tasks).
512#[derive(Clone, Debug, PartialEq)]
513pub struct TaskSummary {
514 /// Task id.
515 pub id: TaskId,
516 /// Key, if any.
517 pub key: Option<Arc<str>>,
518 /// Pool.
519 pub pool: Arc<str>,
520 /// Kind.
521 pub kind: Arc<str>,
522 /// Priority class.
523 pub priority: Priority,
524 /// Current status.
525 pub status: TaskStatus,
526 /// Attempts started so far.
527 pub attempt: u32,
528 /// Tenant (pools with fairness).
529 pub tenant: Option<Arc<str>>,
530}
531
532/// A recurring schedule, as listed by [`Core::list_recurring`](crate::Core::list_recurring).
533#[derive(Clone, Debug, PartialEq)]
534pub struct RecurringInfo {
535 /// Schedule name.
536 pub name: Arc<str>,
537 /// Cron expression.
538 pub cron: Arc<str>,
539 /// Time zone (UTC if `None`).
540 pub tz: Option<Arc<str>>,
541 /// Pool of its occurrences.
542 pub pool: Arc<str>,
543 /// Kind of its occurrences.
544 pub kind: Arc<str>,
545 /// Overlap policy.
546 pub overlap: OverlapPolicy,
547 /// Next occurrence (`None` when the schedule has ended).
548 pub next: Option<Time>,
549 /// Task of the latest occurrence.
550 pub last_task: Option<TaskId>,
551}