sofie_db/worker/
schedules.rs

1use std::sync::Mutex;
2use std::time::Duration;
3
4use sofie_core::{Error, Result, Timestamp, unpoisoned};
5
6use super::Worker;
7use crate::Messages;
8
9/// Queues the schedule `name`'s job of `worker` now, for a worker running beside this process.
10pub async fn queue_schedule(worker: &Worker, name: &str) -> Result<()> {
11    let schedules = &worker.registry.schedules;
12    if !schedules.names().any(|known| known == name) {
13        let mut known: Vec<&str> = schedules.names().collect();
14        known.sort_unstable();
15        let known = if known.is_empty() {
16            "none".to_owned()
17        } else {
18            known.join(", ")
19        };
20        return Err(Error::internal_message(format!(
21            "the app has no schedule `{name}`; its schedules: {known}"
22        )));
23    }
24    worker.messages.queue_schedule(name).await
25}
26
27#[cfg(feature = "local-time")]
28use super::POLL_INTERVAL;
29
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub enum Cadence {
32    Interval(Duration),
33    #[cfg(feature = "local-time")]
34    At(super::WallClock),
35}
36
37impl From<Duration> for Cadence {
38    fn from(interval: Duration) -> Self {
39        Self::Interval(interval)
40    }
41}
42
43#[cfg(feature = "local-time")]
44impl From<super::WallClock> for Cadence {
45    fn from(wall_clock: super::WallClock) -> Self {
46        Self::At(wall_clock)
47    }
48}
49
50impl Cadence {
51    /// An interval schedule first runs as soon as it is new; a wall-clock one at its next time.
52    #[cfg_attr(
53        not(feature = "local-time"),
54        expect(clippy::unnecessary_wraps, reason = "a wall-clock schedule can fail")
55    )]
56    fn first_at(self, now: Timestamp) -> Result<Timestamp> {
57        match self {
58            Self::Interval(_) => Ok(now),
59            #[cfg(feature = "local-time")]
60            Self::At(wall_clock) => wall_clock.next_after(now),
61        }
62    }
63
64    fn unknown_zone(self) -> Option<Error> {
65        match self {
66            Self::Interval(_) => None,
67            #[cfg(feature = "local-time")]
68            Self::At(wall_clock) => wall_clock.zone().err(),
69        }
70    }
71
72    fn key(self) -> String {
73        match self {
74            Self::Interval(interval) => format!("every {}ms", interval.as_millis()),
75            #[cfg(feature = "local-time")]
76            Self::At(wall_clock) => wall_clock.key(),
77        }
78    }
79
80    fn next_after(self, now: Timestamp) -> Result<Timestamp> {
81        match self {
82            Self::Interval(interval) => now
83                .checked_add(interval)
84                .map_err(|error| Error::internal("computing a schedule's next run", error)),
85            #[cfg(feature = "local-time")]
86            Self::At(wall_clock) => wall_clock.next_after(now),
87        }
88    }
89}
90
91#[derive(Default)]
92pub(super) struct Schedules {
93    every: Vec<(&'static str, Cadence, Mutex<Option<Timestamp>>)>,
94}
95
96impl Schedules {
97    pub(super) fn add(&mut self, name: &'static str, cadence: Cadence) {
98        self.every.push((name, cadence, Mutex::new(None)));
99    }
100
101    pub(super) fn names(&self) -> impl Iterator<Item = &'static str> {
102        self.every.iter().map(|(name, ..)| *name)
103    }
104
105    pub(super) fn unknown_zones(&self) -> impl Iterator<Item = String> + '_ {
106        self.every.iter().filter_map(|(name, cadence, _)| {
107            let error = cadence.unknown_zone()?;
108            Some(format!("schedule `{name}` cannot run: {error}"))
109        })
110    }
111
112    pub(super) async fn store(&self, messages: &Messages, now: Timestamp) -> Result<()> {
113        for (name, cadence, _) in &self.every {
114            let first_at = cadence.first_at(now)?;
115            messages
116                .add_schedule(name, first_at, &cadence.key())
117                .await?;
118        }
119        Ok(())
120    }
121
122    /// Another worker may have fired it since; the answer says when it is due next.
123    pub(super) async fn fire_due(&self, messages: &Messages, now: Timestamp) -> Result<()> {
124        for (name, cadence, next_due) in &self.every {
125            let next_due = || unpoisoned::lock(next_due);
126            if next_due().is_some_and(|due| due > now) {
127                continue;
128            }
129            let fired = messages
130                .fire_schedule(name, now, cadence.next_after(now)?)
131                .await?;
132            *next_due() = fired.map(|fired| fired.next_at);
133        }
134        Ok(())
135    }
136}
137
138/// Any process may add a local schedule, so no due time is known; a running worker looks for
139/// them at most once per poll interval instead of before every claim.
140#[cfg(feature = "local-time")]
141#[derive(Default)]
142pub(super) struct LocalLooks(Mutex<Option<Timestamp>>);
143
144#[cfg(feature = "local-time")]
145impl LocalLooks {
146    /// A clock moved back looks again at once.
147    pub(super) fn due(&self, now: Timestamp) -> bool {
148        let mut last = unpoisoned::lock(&self.0);
149        let due = last.is_none_or(|last| {
150            now < last
151                || last
152                    .checked_add(POLL_INTERVAL)
153                    .is_ok_and(|next| now >= next)
154        });
155        if due {
156            *last = Some(now);
157        }
158        due
159    }
160}
161
162#[cfg(all(test, feature = "local-time"))]
163mod tests {
164    use super::*;
165
166    fn at(milliseconds: i64) -> Timestamp {
167        Timestamp::from_millisecond(milliseconds).unwrap()
168    }
169
170    #[test]
171    fn local_schedules_are_looked_for_at_most_once_per_poll_interval() {
172        let looks = LocalLooks::default();
173
174        let seen: Vec<bool> = [0, 10, 999, 1_000, 1_500, 2_000, 1_200]
175            .into_iter()
176            .map(|milliseconds| looks.due(at(milliseconds)))
177            .collect();
178
179        assert_eq!(seen, [true, false, false, true, false, true, true]);
180    }
181}
sofie_db/worker/schedules.rs

sofie_db/worker/
schedules.rs
