headless_lms_server/programs/
periodic_worker.rs

1//! The tick-interval scaffold shared by every background worker that polls the database on a fixed
2//! period: `regrader`, `chatbot_syncer` and the credit registration phase runners.
3
4use std::error::Error as StdError;
5use std::time::Duration;
6
7/// How a worker's ticking loop should behave, so the loop itself carries none of that per-worker
8/// detail.
9pub struct PeriodicWorkerConfig<'a> {
10    pub tick_interval: Duration,
11    /// Ticks between "still running" heartbeat logs.
12    pub still_running_every: u32,
13    pub still_running_message: &'a str,
14    /// Starting value of the tick counter, so a worker that wants its first heartbeat sooner than
15    /// `still_running_every` ticks can seed it.
16    pub initial_ticks: u32,
17    /// `true` pushes a slow iteration's next tick out instead of firing it immediately
18    /// (`tokio::time::MissedTickBehavior::Delay`); `false` keeps tokio's default (`Burst`).
19    pub delay_missed_ticks: bool,
20}
21
22/// Runs `body` on `config.tick_interval` forever, logging `config.still_running_message` every
23/// `config.still_running_every` ticks. A `body` that returns `Err` stops the loop and becomes this
24/// function's return value, the same as an unhandled error used to exit the worker's `main`.
25pub async fn run_periodic_worker(
26    config: PeriodicWorkerConfig<'_>,
27    mut body: impl AsyncFnMut() -> anyhow::Result<()>,
28) -> anyhow::Result<()> {
29    let mut interval = tokio::time::interval(config.tick_interval);
30    if config.delay_missed_ticks {
31        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
32    }
33    let mut ticks = config.initial_ticks;
34    loop {
35        interval.tick().await;
36        ticks += 1;
37        if ticks >= config.still_running_every {
38            ticks = 0;
39            info!("{}", config.still_running_message);
40        }
41        body().await?;
42    }
43}
44
45/// True when an error's source is a `sqlx::Error::Io`, which is usually the database being reset
46/// underneath a local development cluster: the caller's cue to log its own hint and, if it keeps a
47/// connection open across ticks, reacquire one. Takes the source directly (`error.source()`)
48/// rather than the error, since `anyhow::Error` does not implement `std::error::Error`.
49pub fn is_db_disconnect(source: Option<&(dyn StdError + 'static)>) -> bool {
50    matches!(
51        source.and_then(|source| source.downcast_ref::<sqlx::Error>()),
52        Some(sqlx::Error::Io(..))
53    )
54}
headless_lms_server/programs/periodic_worker.rs

headless_lms_server/programs/
periodic_worker.rs
