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 next_after(self, now: Timestamp) -> Result<Timestamp> {
73        match self {
74            Self::Interval(interval) => now
75                .checked_add(interval)
76                .map_err(|error| Error::internal("computing a schedule's next run", error)),
77            #[cfg(feature = "local-time")]
78            Self::At(wall_clock) => wall_clock.next_after(now),
79        }
80    }
81}
82
83#[derive(Default)]
84pub(super) struct Schedules {
85    every: Vec<(&'static str, Cadence, Mutex<Option<Timestamp>>)>,
86}
87
88impl Schedules {
89    pub(super) fn add(&mut self, name: &'static str, cadence: Cadence) {
90        self.every.push((name, cadence, Mutex::new(None)));
91    }
92
93    pub(super) fn names(&self) -> impl Iterator<Item = &'static str> {
94        self.every.iter().map(|(name, ..)| *name)
95    }
96
97    pub(super) fn unknown_zones(&self) -> impl Iterator<Item = String> + '_ {
98        self.every.iter().filter_map(|(name, cadence, _)| {
99            let error = cadence.unknown_zone()?;
100            Some(format!("schedule `{name}` cannot run: {error}"))
101        })
102    }
103
104    pub(super) async fn store(&self, messages: &Messages, now: Timestamp) -> Result<()> {
105        for (name, cadence, _) in &self.every {
106            let first_at = cadence.first_at(now)?;
107            messages
108                .add_schedule(name, first_at, &format!("{cadence:?}"))
109                .await?;
110        }
111        Ok(())
112    }
113
114    /// Another worker may have fired it since; the answer says when it is due next.
115    pub(super) async fn fire_due(&self, messages: &Messages, now: Timestamp) -> Result<()> {
116        for (name, cadence, next_due) in &self.every {
117            let next_due = || unpoisoned::lock(next_due);
118            if next_due().is_some_and(|due| due > now) {
119                continue;
120            }
121            let fired = messages
122                .fire_schedule(name, now, cadence.next_after(now)?)
123                .await?;
124            *next_due() = fired.map(|fired| fired.next_at);
125        }
126        Ok(())
127    }
128}
129
130/// Any process may add a local schedule, so no due time is known; a running worker looks for
131/// them at most once per poll interval instead of before every claim.
132#[cfg(feature = "local-time")]
133#[derive(Default)]
134pub(super) struct LocalLooks(Mutex<Option<Timestamp>>);
135
136#[cfg(feature = "local-time")]
137impl LocalLooks {
138    /// A clock moved back looks again at once.
139    pub(super) fn due(&self, now: Timestamp) -> bool {
140        let mut last = unpoisoned::lock(&self.0);
141        let due = last.is_none_or(|last| {
142            now < last
143                || last
144                    .checked_add(POLL_INTERVAL)
145                    .is_ok_and(|next| now >= next)
146        });
147        if due {
148            *last = Some(now);
149        }
150        due
151    }
152}
153
154#[cfg(all(test, feature = "local-time"))]
155mod tests {
156    use super::*;
157
158    fn at(milliseconds: i64) -> Timestamp {
159        Timestamp::from_millisecond(milliseconds).unwrap()
160    }
161
162    #[test]
163    fn local_schedules_are_looked_for_at_most_once_per_poll_interval() {
164        let looks = LocalLooks::default();
165
166        let seen: Vec<bool> = [0, 10, 999, 1_000, 1_500, 2_000, 1_200]
167            .into_iter()
168            .map(|milliseconds| looks.due(at(milliseconds)))
169            .collect();
170
171        assert_eq!(seen, [true, false, false, true, false, true, true]);
172    }
173}
sofie_db/worker/schedules.rs

sofie_db/worker/
schedules.rs
