rustlavel_queue/
chain.rs

1//! Jobs that run one after another.
2//!
3//! A pipeline — transcribe, then detect the moments, then render each — is a
4//! chain: the next step is dispatched when the one before it succeeds, and a
5//! step that fails for good ends the chain there. The steps are ordinary jobs;
6//! nothing about a job knows it is in a chain except the context it is handed,
7//! which carries the chain's id so progress can be reported against the whole.
8//!
9//! ```ignore
10//! queue.dispatch_chain(
11//!     Chain::new(Transcribe { video })
12//!         .then(DetectMoments { video })
13//!         .then(RenderClips { video }),
14//! ).await?;
15//! ```
16//!
17//! **A failed step stops the chain**, and the remaining steps go to the
18//! dead-letter table with it, so what did not run is visible rather than
19//! silently gone. Retries happen within the step, with the step's own policy;
20//! the chain only advances on success.
21
22use crate::job::{Job, QueuedJob};
23
24/// A sequence of jobs.
25#[derive(Debug, Clone, PartialEq)]
26pub struct Chain {
27    /// A name for the whole, so progress for the pipeline has one key.
28    /// Generated when not given.
29    pub id: String,
30    steps: Vec<QueuedJob>,
31}
32
33impl Chain {
34    /// Start a chain with its first step.
35    pub fn new<J: Job>(first: J) -> Chain {
36        Chain { id: chain_id(), steps: vec![first.to_queued()] }
37    }
38
39    /// Name the chain yourself — an order id, a video id — so the key a UI
40    /// watches is one it already knows.
41    pub fn named(mut self, id: impl Into<String>) -> Chain {
42        self.id = id.into();
43        self
44    }
45
46    pub fn then<J: Job>(mut self, next: J) -> Chain {
47        self.steps.push(next.to_queued());
48        self
49    }
50
51    /// Add a step that is already an envelope, for a job with no Rust type.
52    pub fn then_queued(mut self, next: QueuedJob) -> Chain {
53        self.steps.push(next);
54        self
55    }
56
57    pub fn len(&self) -> usize {
58        self.steps.len()
59    }
60
61    pub fn is_empty(&self) -> bool {
62        self.steps.is_empty()
63    }
64
65    /// The first step, carrying the rest and the chain's id. This is what is
66    /// pushed; the worker peels one step off each time one succeeds.
67    pub fn into_queued(mut self) -> QueuedJob {
68        let mut first = self.steps.remove(0);
69        first.chain = Some(self.id);
70        first.then = self.steps;
71        first
72    }
73}
74
75fn chain_id() -> String {
76    // Time and a counter: unique within a process and very likely across
77    // them, and readable in a log. Not a secret, so not from the CSPRNG.
78    static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
79    let count = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
80    format!("{:x}-{count:x}", crate::time::unix_now())
81}
82
83impl QueuedJob {
84    /// The job to push once this one has succeeded, if this one is a step in
85    /// a chain with more to come. The returned job carries the remainder.
86    pub fn next_in_chain(&self) -> Option<QueuedJob> {
87        let mut rest = self.then.clone();
88        if rest.is_empty() {
89            return None;
90        }
91        let mut next = rest.remove(0);
92        next.chain = self.chain.clone();
93        next.then = rest;
94        Some(next)
95    }
96}
rustlavel_queue/chain.rs

rustlavel_queue/
chain.rs
