sofie_db/contracts/
local.rs

1//! Jobs at a wall-clock time in a zone, such as 09:00 in Europe/Istanbul every day.
2//! Kept as that local time, so each run is placed by the zone's rules as they are then.
3
4use std::sync::Arc;
5use std::time::Duration;
6
7use serde::Serialize;
8use serde_json::Value;
9use sofie_core::variants::{self, Variants};
10use sofie_core::{
11    Cast, ColumnReader, ColumnSpec, Error, Result, SqlType, SqlValue, Timestamp, Uuid,
12};
13use sofie_values::LocalDateTime;
14
15use super::jobs::{JobNames, Jobs, Queued};
16use crate::Resolve;
17use crate::worker::JobContext;
18use crate::{
19    Aggregate, Call, Column, Db, Field, Id, Load, MessageKind, Meta, Outcome, Plan, Reach, Select,
20    Stamp, Writes,
21};
22
23const TEXT: &[ColumnSpec] = &[ColumnSpec::new("", SqlType::Text)];
24const DAY: Duration = Duration::from_hours(24);
25/// Fired per pass, so a backlog never holds the worker's claim for long.
26const BATCH: u64 = 100;
27/// A schedule that cannot fire waits this long, so it neither floods the log nor fills each batch.
28const FAILED_RETRY_AFTER: Duration = Duration::from_mins(15);
29
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub enum Repeat {
32    Once,
33    Daily,
34    Weekly,
35}
36
37impl Variants for Repeat {
38    const TYPE: &'static str = "local schedule repeat";
39    const VARIANTS: &'static [(&'static str, Self)] = &[
40        ("once", Self::Once),
41        ("daily", Self::Daily),
42        ("weekly", Self::Weekly),
43    ];
44}
45
46impl Repeat {
47    fn step(self) -> Option<Duration> {
48        match self {
49            Self::Once => None,
50            Self::Daily => Some(DAY),
51            Self::Weekly => Some(DAY * 7),
52        }
53    }
54}
55
56pub(crate) struct LocalSchedule {
57    meta: Meta,
58    key: String,
59    job: String,
60    payload: String,
61    at: LocalDateTime,
62    repeat: Repeat,
63    /// `None` once a single run ran, or when canceled.
64    next_at: Option<Timestamp>,
65    owner: Option<Uuid>,
66}
67
68const KEY: Column<LocalSchedule, String> = Column::new("key");
69const NEXT_AT: Column<LocalSchedule, Timestamp> = Column::new("next_at");
70
71/// Sets and cancels jobs at a local time, each under a key the app chooses, such as
72/// `digest:<account id>`, so setting it again moves it.
73pub struct LocalJobs {
74    jobs: Jobs,
75    db: Db,
76    owner: Option<Uuid>,
77}
78
79impl<C: Writes> Resolve<C> for LocalJobs {
80    fn resolve(context: &mut C) -> impl Future<Output = Result<Self>> + Send {
81        std::future::ready(Ok(Self::within(context.db(), context.reach())))
82    }
83}
84
85impl LocalJobs {
86    /// For a package resolving its own arguments: what `reach` owns is what it writes.
87    #[must_use]
88    pub fn within(db: &Db, reach: Reach) -> Self {
89        let names = db.provided.get::<JobNames>().unwrap_or_default();
90        Self {
91            jobs: Jobs(names),
92            db: db.clone(),
93            owner: reach.owner(),
94        }
95    }
96
97    /// The first run is the next time `at` falls at or after now; a single one already past runs at once.
98    pub async fn set<Args, J: Call<JobContext, Args>>(
99        &self,
100        key: &str,
101        job: J,
102        payload: &impl Serialize,
103        at: LocalDateTime,
104        repeat: Repeat,
105    ) -> Result<LocalScheduled> {
106        let Queued {
107            name: job, payload, ..
108        } = self.jobs.queue(job, payload)?;
109        let now = self.db.clock().now().timestamp();
110        let (at, next_at) = first_run(at, repeat, now)?;
111        let schedule = match self.keyed(key).await? {
112            Some(mut schedule) => {
113                job.clone_into(&mut schedule.job);
114                schedule.payload = payload.to_string();
115                schedule.at = at;
116                schedule.repeat = repeat;
117                schedule.next_at = Some(next_at);
118                schedule
119            }
120            None => LocalSchedule {
121                meta: Meta::new()?,
122                key: key.to_owned(),
123                job: job.to_owned(),
124                payload: payload.to_string(),
125                at,
126                repeat,
127                next_at: Some(next_at),
128                owner: self.owner,
129            },
130        };
131        Ok(LocalScheduled(Some(schedule)))
132    }
133
134    /// Nothing changes when no schedule has the key.
135    pub async fn cancel(&self, key: &str) -> Result<LocalScheduled> {
136        let schedule = self.keyed(key).await?.map(|mut schedule| {
137            schedule.next_at = None;
138            schedule
139        });
140        Ok(LocalScheduled(schedule))
141    }
142
143    async fn keyed(&self, key: &str) -> Result<Option<LocalSchedule>> {
144        LocalSchedule::query()
145            .where_(KEY.eq(key.to_owned()))
146            .first(&self.db.read())
147            .await
148    }
149}
150
151#[must_use = "the schedule changes only when the handler returns it"]
152pub struct LocalScheduled(Option<LocalSchedule>);
153
154impl LocalScheduled {
155    #[must_use]
156    pub fn next_at(&self) -> Option<Timestamp> {
157        self.0.as_ref().and_then(|schedule| schedule.next_at)
158    }
159}
160
161impl Outcome for LocalScheduled {
162    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
163        self.0
164            .as_ref()
165            .map_or(Ok(()), |schedule| schedule.plan(plan))
166    }
167
168    fn committed(&mut self, stamp: Stamp) {
169        if let Some(schedule) = &mut self.0 {
170            schedule.committed(stamp);
171        }
172    }
173}
174
175/// For a repeating run, the occurrence after `now` at the earliest, counted in the zone's days.
176fn first_run(
177    at: LocalDateTime,
178    repeat: Repeat,
179    now: Timestamp,
180) -> Result<(LocalDateTime, Timestamp)> {
181    let instant = at.instant()?.timestamp();
182    let Some(step) = repeat.step() else {
183        return Ok((at, instant));
184    };
185    if instant >= now {
186        return Ok((at, instant));
187    }
188    following(&at, step, now)
189}
190
191/// Skips runs a stopped worker missed rather than running each of them late.
192fn following(
193    at: &LocalDateTime,
194    step: Duration,
195    now: Timestamp,
196) -> Result<(LocalDateTime, Timestamp)> {
197    let behind = u64::try_from((now.as_second() - at.instant()?.timestamp().as_second()).max(0))
198        .unwrap_or(0);
199    let skipped = (behind / step.as_secs()).saturating_sub(1);
200    let mut local = advance(at, step * u32::try_from(skipped).unwrap_or(u32::MAX))?;
201    loop {
202        let instant = local.instant()?.timestamp();
203        if instant > now {
204            return Ok((local, instant));
205        }
206        local = advance(&local, step)?;
207    }
208}
209
210fn advance(at: &LocalDateTime, by: Duration) -> Result<LocalDateTime> {
211    let local = at
212        .local()
213        .checked_add(by)
214        .map_err(|error| Error::internal("moving a local schedule on", error))?;
215    Ok(LocalDateTime::new(local, at.zone().clone()))
216}
217
218/// A worker that fired a schedule first wins the version check, so a run is queued once.
219pub(crate) async fn fire_due(db: &Db, now: Timestamp) -> Result<()> {
220    let due = LocalSchedule::query()
221        .where_(NEXT_AT.lte(now))
222        .limit(BATCH)
223        .all(&db.read())
224        .await?;
225    for schedule in due {
226        let (id, key) = (schedule.id(), schedule.key.clone());
227        let Err(error) = fire(db, schedule, now).await else {
228            continue;
229        };
230        if matches!(error, Error::Conflict) {
231            continue;
232        }
233        tracing::error!(schedule = %key, error = %error, "firing a local schedule failed");
234        if let Err(error) = put_off(db, id, now).await {
235            tracing::error!(schedule = %key, error = %error, "putting off a local schedule failed");
236        }
237    }
238    Ok(())
239}
240
241async fn put_off(db: &Db, id: Id<LocalSchedule>, now: Timestamp) -> Result<()> {
242    let mut schedule = LocalSchedule::find(&db.read(), id).await?;
243    schedule.next_at = Some(
244        now.checked_add(FAILED_RETRY_AFTER)
245            .map_err(|error| Error::internal("putting off a local schedule", error))?,
246    );
247    db.commit(&mut schedule).await
248}
249
250async fn fire(db: &Db, mut schedule: LocalSchedule, now: Timestamp) -> Result<()> {
251    // The zone's rules may have moved the time since the run was placed.
252    let placed = schedule.at.instant()?.timestamp();
253    if placed > now {
254        schedule.next_at = Some(placed);
255        return db.commit(&mut schedule).await;
256    }
257    let payload: Value = serde_json::from_str(&schedule.payload)
258        .map_err(|error| Error::internal("reading a local schedule's payload", error))?;
259    match schedule.repeat.step() {
260        None => schedule.next_at = None,
261        Some(step) => {
262            let (at, next_at) = following(&schedule.at, step, now)?;
263            schedule.at = at;
264            schedule.next_at = Some(next_at);
265        }
266    }
267    let job: Arc<str> = Arc::from(schedule.job.as_str());
268    db.commit(&mut Fired {
269        schedule,
270        job,
271        payload,
272    })
273    .await
274}
275
276struct Fired {
277    schedule: LocalSchedule,
278    job: Arc<str>,
279    payload: Value,
280}
281
282impl Outcome for Fired {
283    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
284        self.schedule.plan(plan)?;
285        plan.queue(MessageKind::Job, [(&*self.job, self.payload.clone())])
286    }
287
288    fn committed(&mut self, stamp: Stamp) {
289        self.schedule.committed(stamp);
290    }
291}
292
293impl Aggregate for LocalSchedule {
294    const NAME: &'static str = "local_schedules";
295    const MODULE: &'static str = "sofie";
296    const FIELDS: &'static [Field] = &[
297        Field::new("key", String::COLUMNS, false),
298        Field::new("job", String::COLUMNS, false),
299        Field::new("payload", TEXT, false),
300        Field::new("at", LocalDateTime::COLUMNS, false),
301        Field::new("repeat", variants::COLUMNS, false),
302        Field::new("next_at", Timestamp::COLUMNS, true),
303        Field::new("owner", Uuid::COLUMNS, true),
304    ];
305    const UNIQUE: &'static [&'static str] = &["key"];
306
307    fn meta(&self) -> &Meta {
308        &self.meta
309    }
310
311    fn meta_mut(&mut self) -> &mut Meta {
312        &mut self.meta
313    }
314
315    fn owner(&self) -> Option<Uuid> {
316        self.owner
317    }
318
319    fn encode(&self, out: &mut Vec<SqlValue>) {
320        self.key.encode(out);
321        self.job.encode(out);
322        out.push(SqlValue::Text(self.payload.clone()));
323        self.at.encode(out);
324        variants::encode(self.repeat, out);
325        self.next_at.encode(out);
326        self.owner.encode(out);
327    }
328
329    fn decode(meta: Meta, row: &mut dyn ColumnReader) -> Result<Self> {
330        Ok(Self {
331            meta,
332            key: String::decode(row)?,
333            job: String::decode(row)?,
334            payload: row.text()?.to_owned(),
335            at: LocalDateTime::decode(row)?,
336            repeat: variants::decode(row)?,
337            next_at: Option::decode(row)?,
338            owner: Option::decode(row)?,
339        })
340    }
341}
342
343#[cfg(test)]
344mod tests {
345    use sofie_values::TimeZone;
346
347    use super::*;
348
349    fn local(text: &str) -> LocalDateTime {
350        LocalDateTime::parse(text, TimeZone::parse("Europe/Berlin").unwrap()).unwrap()
351    }
352
353    fn at(text: &str) -> Timestamp {
354        text.parse().unwrap()
355    }
356
357    #[test]
358    fn a_daily_run_keeps_its_wall_clock_time_across_a_shift() {
359        let nine = local("2026-10-24T09:00");
360
361        let (next, instant) = following(&nine, DAY, at("2026-10-24T08:00:00Z")).unwrap();
362
363        assert_eq!(next.local().to_string(), "2026-10-25T09:00:00");
364        assert_eq!(instant, at("2026-10-25T08:00:00Z"));
365    }
366
367    #[test]
368    fn a_skipped_time_runs_where_the_shift_ends() {
369        let half_past_two = local("2026-03-28T02:30");
370
371        let (next, instant) = following(&half_past_two, DAY, at("2026-03-28T12:00:00Z")).unwrap();
372
373        assert_eq!(next.local().to_string(), "2026-03-29T02:30:00");
374        assert_eq!(instant, at("2026-03-29T01:00:00Z"));
375    }
376
377    #[test]
378    fn a_repeating_run_set_in_the_past_starts_at_its_next_time_and_a_single_one_at_once() {
379        let now = at("2026-10-14T12:00:00Z");
380        let long_ago = local("2025-01-06T09:00");
381
382        let (weekly, _) = first_run(long_ago.clone(), Repeat::Weekly, now).unwrap();
383        let (once, instant) = first_run(long_ago, Repeat::Once, now).unwrap();
384
385        assert_eq!(weekly.local().to_string(), "2026-10-19T09:00:00");
386        assert_eq!(once.local().to_string(), "2025-01-06T09:00:00");
387        assert!(instant < now);
388    }
389}
sofie_db/contracts/local.rs

sofie_db/contracts/
local.rs
