sofie_db/worker/
mod.rs

1mod backoff;
2mod context;
3mod each;
4#[cfg(feature = "remote-contracts")]
5mod feeds;
6mod flight;
7mod handling;
8#[cfg(feature = "jobs")]
9mod jobs;
10mod lease;
11#[cfg(feature = "metrics")]
12mod metrics;
13mod registry;
14mod schedules;
15mod trace;
16mod wake;
17#[cfg(feature = "local-time")]
18mod wall_clock;
19
20use std::num::{NonZeroU16, NonZeroUsize};
21use std::pin::pin;
22use std::time::Duration;
23
24use sofie_core::Result;
25
26pub use context::JobContext;
27#[cfg(feature = "remote-contracts")]
28pub use feeds::Feeds;
29pub use registry::publish_heard;
30pub use schedules::{Cadence, queue_schedule};
31#[cfg(feature = "local-time")]
32pub use wall_clock::WallClock;
33
34use crate::{Aggregate, Call, Db, Entry, Messages, Run, Trigger, erase};
35pub(crate) use registry::Registry;
36
37use wake::Wake;
38
39const DEFAULT_MAX_ATTEMPTS: i64 = 10;
40/// Retries and schedules come due without a notification.
41const POLL_INTERVAL: Duration = Duration::from_secs(1);
42/// Renewed while its job runs; a worker that dies frees its messages after it.
43const LEASE: Duration = Duration::from_mins(5);
44/// Below the lease, so a job whose renewals fail ends before another worker may claim it.
45const JOB_TIMEOUT: Duration = Duration::from_mins(4);
46/// A message no worker here handles may be a newer version's, queued during a rolling deploy.
47const UNHEARD_KEPT_FOR: Duration = Duration::from_hours(1);
48/// Loco's default worker count, so both frameworks start a job queue the same way.
49const DEFAULT_IN_FLIGHT: usize = 2;
50
51/// Any number of workers can share a database; each message is handled by one.
52pub struct Worker {
53    db: Db,
54    messages: Messages,
55    registry: Registry,
56    max_attempts: i64,
57    in_flight: usize,
58    #[cfg(feature = "local-time")]
59    local_looks: schedules::LocalLooks,
60    #[cfg(feature = "remote-contracts")]
61    feeds: Feeds,
62}
63
64impl Worker {
65    #[must_use]
66    pub fn new(db: Db) -> Self {
67        Self {
68            messages: Messages::new(db.clone()),
69            registry: Registry::default(),
70            db,
71            max_attempts: DEFAULT_MAX_ATTEMPTS,
72            in_flight: DEFAULT_IN_FLIGHT,
73            #[cfg(feature = "local-time")]
74            local_looks: schedules::LocalLooks::default(),
75            #[cfg(feature = "remote-contracts")]
76            feeds: Feeds::default(),
77        }
78    }
79
80    /// Runs `job` for each `E` raised.
81    #[must_use]
82    pub fn on<E: crate::Event, Args: 'static>(
83        self,
84        name: &'static str,
85        job: impl Call<JobContext, Args>,
86    ) -> Self {
87        self.register(name, Trigger::On(E::NAME), erase(job))
88    }
89
90    /// At an interval, first as soon as the schedule is new; or at a `WallClock` time, first
91    /// when the zone's clocks next show it.
92    #[must_use]
93    pub fn every<Args: 'static, J: Call<JobContext, Args>>(
94        self,
95        name: &'static str,
96        cadence: impl Into<Cadence>,
97        job: J,
98    ) -> Self {
99        self.register(name, Trigger::Every(cadence.into()), erase(job))
100    }
101
102    /// Each record's job runs with its own retries.
103    #[must_use]
104    pub fn each<A: Aggregate, QueryArgs: 'static, Args: 'static>(
105        self,
106        name: &'static str,
107        interval: Duration,
108        items: impl Call<JobContext, QueryArgs, Outcome = Vec<A>>,
109        action: &'static str,
110        job: impl Call<JobContext, Args>,
111    ) -> Self {
112        self.register(action, Trigger::Queued, erase(job)).register(
113            name,
114            Trigger::Every(Cadence::Interval(interval)),
115            each::each(items, action),
116        )
117    }
118
119    fn register(mut self, name: &'static str, trigger: Trigger, run: Run<JobContext>) -> Self {
120        self.registry.add(Entry { name, trigger, run });
121        self
122    }
123
124    /// A new mirror first copies every record its source has, once.
125    #[cfg(feature = "contracts")]
126    #[must_use]
127    pub fn mirror<M: crate::Mirror>(self, name: &'static str) -> Self {
128        const ONCE: Duration = Duration::from_hours(100 * 365 * 24);
129        let backfill: &'static str = Box::leak(format!("{name}.backfill").into_boxed_str());
130        let copy = move |source: crate::Dep<dyn crate::Lookup<M::Source>>, page| {
131            crate::backfill_mirror::<M>(source, page, backfill)
132        };
133        let keep = |row: crate::MirrorRow<M>| async move { Ok(row) };
134        let mut worker = self.every(backfill, ONCE, copy);
135        worker.registry.add_taking_pruned(Entry {
136            name,
137            trigger: Trigger::On(<M::Changed as crate::Event>::NAME),
138            run: erase(keep),
139        });
140        if worker
141            .registry
142            .job(crate::contracts::SWEEP_FORGOTTEN)
143            .is_some()
144        {
145            return worker;
146        }
147        worker.every(
148            crate::contracts::SWEEP_FORGOTTEN,
149            Duration::from_hours(1),
150            crate::contracts::sweep_forgotten,
151        )
152    }
153
154    /// At least one, as none would leave every message unhandled:
155    ///
156    /// ```compile_fail,E0308
157    /// fn never(worker: sofie_db::worker::Worker) -> sofie_db::worker::Worker {
158    ///     worker.max_attempts(0)
159    /// }
160    /// ```
161    #[must_use]
162    pub fn max_attempts(mut self, max_attempts: NonZeroU16) -> Self {
163        self.max_attempts = i64::from(max_attempts.get());
164        self
165    }
166
167    #[must_use]
168    pub fn batch(mut self, size: NonZeroUsize) -> Self {
169        self.in_flight = size.get();
170        self
171    }
172
173    /// Refuses repeated job names, intervals out of range and an unkept feed before any work.
174    pub async fn run(self, shutdown: impl Future<Output = ()> + Send) -> Result<()> {
175        self.registry.validate()?;
176        #[cfg(feature = "remote-contracts")]
177        self.feeds.refuse_unkept(&self.db)?;
178        publish_heard(&self);
179        self.registry
180            .schedules
181            .store(&self.messages, self.db.clock().now().timestamp())
182            .await?;
183        #[cfg(feature = "metrics")]
184        crate::metrics::watch_outbox(&self.db);
185        #[cfg(feature = "remote-contracts")]
186        let (shutdown, pulling) = self.feeds.start(&self.db, shutdown);
187        let wake = &mut Wake::new(self.messages.clone());
188        let worked = flight::work(&self, Some((pin!(shutdown), wake))).await;
189        #[cfg(feature = "remote-contracts")]
190        pulling.finish().await?;
191        worked.map(drop)
192    }
193
194    /// Pulls every feed first; a failed job is not counted, it is retried once its backoff is over.
195    pub async fn drain(&self) -> Result<usize> {
196        self.drain_reporting()
197            .await
198            .map(|drained| drained.completed)
199    }
200
201    pub async fn drain_reporting(&self) -> Result<Drained> {
202        self.registry.validate()?;
203        #[cfg(feature = "remote-contracts")]
204        self.feeds.refuse_unkept(&self.db)?;
205        publish_heard(self);
206        self.registry
207            .schedules
208            .store(&self.messages, self.db.clock().now().timestamp())
209            .await?;
210        #[cfg(feature = "remote-contracts")]
211        self.feeds.pull_all(&self.db).await?;
212        flight::work(self, None).await
213    }
214}
215
216#[derive(Debug, Default)]
217pub struct Drained {
218    pub completed: usize,
219    pub failures: Vec<Failure>,
220}
221
222/// A panic counts as a failure too.
223#[derive(Clone, Debug, PartialEq, Eq)]
224pub struct Failure {
225    pub job: String,
226    pub error: String,
227}
sofie_db/worker/mod.rs

sofie_db/worker/
mod.rs
